Я экспериментирую с Gradient Boosted Trees алгоритмом обучения из библиотеки ML от Spark 1.4. Я решаю проблему с двоичной классификацией, где мой вход составляет ~ 50 000 выборок и ~ 500 000 функций. Моя цель - вывести определение результирующего ансамбля GBT в формате для чтения человеком. Мой опыт до сих пор заключается в том, что для моего размера проблемы добавление большего количества ресурсов в кластер, похоже, не влияет на продолжительность прогона. Кажется, что 10-итерационный тренировочный пробег занимает 13 часов. Это неприемлемо, так как я пытаюсь выполнить 100-300 итераций, и время выполнения, похоже, взорвется с количеством итераций.

Приложение My Spark

Это не точный код, но его можно свести к:

SparkConf sc = new SparkConf().setAppName("GBT Trainer")
            // unlimited max result size for intermediate Map-Reduce ops.
            // Having no limit is probably bad, but I've not had time to find
            // a tighter upper bound and the default value wasn't sufficient.
            .set("spark.driver.maxResultSize", "0");
JavaSparkContext jsc = new JavaSparkContext(sc)

// The input file is encoded in plain-text LIBSVM format ~59GB in size
<LabeledPoint> data = MLUtils.loadLibSVMFile(jsc.sc(), "s3://somebucket/somekey/plaintext_libsvm_file").toJavaRDD();

BoostingStrategy boostingStrategy = BoostingStrategy.defaultParams("Classification");
boostingStrategy.setNumIterations(10);
boostingStrategy.getTreeStrategy().setNumClasses(2);
boostingStrategy.getTreeStrategy().setMaxDepth(1);
Map<Integer, Integer> categoricalFeaturesInfo = new HashMap<Integer, Integer>();
boostingStrategy.treeStrategy().setCategoricalFeaturesInfo(categoricalFeaturesInfo);

GradientBoostedTreesModel model = GradientBoostedTrees.train(data, boostingStrategy);

// Somewhat-convoluted code below reads in Parquete-formatted output
// of the GBT model and writes it back out as json.
// There might be cleaner ways of achieving the same, but since output
// size is only a few KB I feel little guilt leaving it as is.

// serialize and output the GBT classifier model the only way that the library allows
String outputPath = "s3://somebucket/somekeyprefex";
model.save(jsc.sc(), outputPath + "/parquet");
// read in the parquet-formatted classifier output as a generic DataFrame object
SQLContext sqlContext = new SQLContext(jsc);
DataFrame outputDataFrame = sqlContext.read().parquet(outputPath + "/parquet"));    
// output DataFrame-formatted classifier model as json           
outputDataFrame.write().format("json").save(outputPath + "/json");

Вопрос

Каково узкое место производительности моего приложения Spark (или самого алгоритма обучения GBT) на входе этого размера и как я могу добиться большего выполнения parallelism?

Я все еще новичок Spark dev, и я был бы признателен за любые советы по конфигурации кластера и профилированию выполнения.

Подробнее о настройке кластера

Я запускаю это приложение в кластере AWS EMR (emr-4.0.0, режим кластеров YARN) экземпляров r3.8xlarge (32 ядра, каждая по 244 ГБ). Я использую такие большие экземпляры, чтобы максимизировать гибкость распределения ресурсов. До сих пор я пытался использовать 1-3 r3.8x больших экземпляров с различными схемами распределения ресурсов между драйвером и рабочими. Например, для кластера из 1 r3.8x больших экземпляров я отправляю приложение следующим образом:

aws emr add-steps --cluster-id $1 --steps Name=$2,\
Jar=s3://us-east-1.elasticmapreduce/libs/script-runner/script-runner.jar,\
Args=[/usr/lib/spark/bin/spark-submit,--verbose,\
--deploy-mode,cluster,--master,yarn,\
--driver-memory,60G,\
--executor-memory,30G,\
--executor-cores,5,\
--num-executors,6,\
--class,GbtTrainer,\
"s3://somebucket/somekey/spark.jar"],\
ActionOnFailure=CONTINUE

Для кластера из 3-х кратных экземпляров я увеличиваю выделение ресурсов:

--driver-memory,80G,\
--executor-memory,35G,\
--executor-cores,5,\
--num-executors,18,\

У меня нет четкого представления о том, сколько памяти полезно придать каждому исполнителю, но я чувствую, что в любом случае я щедр. Просматривая интерфейс Spark, я не вижу задачи с размером ввода более нескольких ГБ. Я руляю на стороне осторожности, давая драйверу столько памяти, чтобы убедиться, что это не память, голодная для любых промежуточных операций агрегации результатов.

Я пытаюсь сохранить количество ядер на каждом исполнителе до 5 в соответствии с предложениями в Cloudera Как настроить серию Spark Jobs (по их мнению, больше что 5 ядер имеют тенденцию вводить узкое место HDFS IO). Я также убеждаюсь, что осталось достаточно запасных ОЗУ и процессоров для ОС хоста и Hadoop.

Мои результаты до сих пор

Моя единственная подсказка - это Spark UI, показывающий очень длинную Запланирующую задержку для ряда задач в конце выполнения. У меня также возникает ощущение, что временная шкала этапов/задач, отображаемая пользовательским интерфейсом Spark, не учитывает все время, которое требуется для завершения работы. Я подозреваю, что приложение драйвера застревает, выполняя какую-то длительную операцию либо в конце каждой итерации обучения, либо в конце всего тренировочного прогона.

Я уже провел справедливое исследование по настройке приложений Spark. В большинстве статей приводятся отличные рекомендации по использованию RDD-операций, которые уменьшают размер промежуточного ввода или избегают перетасовки данных между этапами. В моем случае я в основном использую алгоритм "из коробки", который написан экспертами ML и должен быть хорошо настроен в этом отношении. Мой собственный код, который выводит модель GBT на S3, должен занимать тривиальное количество времени для запуска.

Медленная производительность с использованием Apache Spark Gradient Boosted Tree

Приложение My Spark

Вопрос

Подробнее о настройке кластера

Мои результаты до сих пор

Ответ 1