amazon-web-services - Apache Spark Gradient Boosted Tree 训练运行性能缓慢

Question

我正在尝试 Gradient Boosted Trees从Spark 1.4的ML库学习算法。我正在解决一个二元分类问题，其中我的输入是 ~50,000 个样本和 ~500,000 个特征。我的目标是以人类可读的格式输出生成的 GBT 集合的定义。到目前为止，我的经验是，对于我的问题大小，向集群添加更多资源似乎不会对运行时间产生影响。 10 次迭代的训练运行似乎大约需要 13 小时。这是 Not Acceptable ，因为我希望进行 100-300 次迭代运行，并且执行时间似乎随着迭代次数的增加而激增。

我的 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 学习算法本身)在该大小的输入上的性能瓶颈是什么？如何实现更大的执行并行度？

我仍然是 Spark 开发新手，如果有关于集群配置和执行分析的提示，我将不胜感激。

有关集群设置的更多详细信息

我在 r3.8xlarge 实例(32 个核心，每个 244GB RAM)的 AWS EMR 集群(emr-4.0.0，YARN 集群模式)上运行此应用程序。我使用如此大的实例是为了最大限度地提高资源分配的灵 active 。到目前为止，我已经尝试使用 1-3 个 r3.8xlarge 实例以及驱动程序和工作人员之间的各种资源分配方案。例如，对于包含 1 个 r3.8xlarge 实例的集群，我按如下方式提交应用程序:

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 个 r3.8xlarge 实例的集群，我调整资源分配:

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

我不清楚为每个执行者提供多少内存是有用的，但我觉得无论哪种情况我都很慷慨。浏览 Spark UI，我没有看到输入大小超过几 GB 的任务。在为驱动程序进程提供如此多的内存时，我会谨慎行事，以确保它不会因任何中间结果聚合操作而内存不足。

按照 Cloudera's How To Tune Your Spark Jobs series 中的建议，我尝试将每个执行器的核心数量保持在 5 个以内。 (根据他们的说法，超过 5 个核心往往会引入 HDFS IO 瓶颈)。我还要确保有足够的备用 RAM 和 CPU 用于主机操作系统和 Hadoop 服务。

到目前为止我的发现

我唯一的线索是 Spark UI 在执行尾部显示许多任务的调度延迟非常长。我还感觉 Spark UI 显示的阶段/任务时间线并没有考虑到完成作业所需的所有时间。我怀疑驱动程序应用程序在每次训练迭代结束时或在整个训练运行结束时卡住了执行某种冗长的操作。

我已经对 Spark 应用程序的调优进行了大量研究。大多数文章都会给出关于使用 RDD 操作的很好的建议，这些操作可以减少中间输入的大小或避免阶段之间的数据混洗。就我而言，我基本上使用的是“开箱即用”算法，该算法是由机器学习专家编写的，并且应该在这方面已经进行了很好的调整。我自己的将 GBT 模型输出到 S3 的代码应该只需要很短的时间即可运行。

Answer 1

我没有使用过 MLLibs GBT 实现，但我都使用过

LightGBM和 XGBoost成功地。我强烈建议您看看这些其他库。

一般来说，GBM 实现需要迭代地训练模型，因为它们在构建下一棵树时考虑到整个集成的损失。这使得 GBM 训练本质上存在瓶颈，并且不容易并行化(与可以轻松并行化的随机森林不同)。我希望它能在任务较少的情况下表现更好，但这可能不是您的全部问题。由于您有如此之多的特征(500K)，因此在训练期间计算直方图和分割点时，您将有非常高的开销。您应该减少拥有的特征数量，特别是因为它们比样本数量大得多，这会导致过度拟合。

至于调整集群:您希望最大限度地减少数据移动，从而减少执行程序并获得更多内存。每个 ec2 实例 1 个执行程序，核心数量设置为实例提供的任何数量。

您的数据足够小，可以容纳大约 2 个该大小的 EC2。假设您使用 double (8 字节)，则大小为 8 * 500000 * 50000 = 200 GB 尝试在数据帧上使用 .cache() 将其全部加载到 RAM 中。如果您对所有行执行操作(例如求和)，您应该强制加载它，并且可以测量 IO 花费的时间。一旦它进入内存并缓存任何其他操作，它就会更快。