Spark ML 梯度提升树不使用所有节点

Question

我正在使用 pyspark 中的 Spark ML GBTClassifier 在 AWS EMR 集群上具有约 400k 行和约 9k 列的数据帧上训练二进制分类模型。我正在将此与我当前的解决方案进行比较，该解决方案在一个巨大的 EC2 上运行 XGBoost，可以将整个数据帧放入内存中。

我希望我可以在 Spark 中更快地训练（并获得新的观察结果），因为它是分布式/并行的。然而，当观察我的集群时（通过神经节），我看到只有 3-4 个节点有活动的 CPU，而其余的节点只是坐在那里。事实上，从它的外观来看，它可能只使用了一个节点进行实际训练。

我似乎在文档中找不到任何关于节点限制或分区的内容，或者任何似乎与为什么会发生这种情况相关的内容。也许我只是误解了算法的实现，但我认为它的实现方式可以并行化训练以利用 Spark 的 EMR/集群方面。如果不是，那么这样做与仅在单个 EC2 上的内存中进行比较有什么优势吗？我想您不必将数据加载到内存中，但这并不是什么优势。

这是我的代码的一些样板。感谢您的任何想法！

import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import DoubleType
from pyspark.ml.classification import GBTClassifier
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# Start Spark context:
sc = pyspark.SparkContext()
sqlContext = SparkSession.builder.enableHiveSupport().getOrCreate()

# load data
df = sqlContext.sql('SELECT label, features FROM full_table WHERE train = 1')
df.cache()
print("training data loaded: {} rows".format(df.count()))

test_df = sqlContext.sql('SELECT label, features FROM full_table WHERE train = 0')
test_df.cache()
print("test data loaded: {} rows".format(test_df.count()))


#Create evaluator
evaluator = BinaryClassificationEvaluator()
evaluator.setRawPredictionCol('prob')
evaluator.setLabelCol('label')

# train model
gbt = GBTClassifier(maxIter=100, 
                    maxDepth=3, 
                    stepSize=0.1,
                    labelCol="label", 
                    seed=42)

model = gbt.fit(df)


# get predictions
gbt_preds = model.transform(test_df)
gbt_preds.show(10)


# evaluate predictions
getprob=udf(lambda v:float(v[1]),DoubleType())
preds = gbt_preds.withColumn('prob', getprob('probability'))\
        .drop('features', 'rawPrediction', 'probability', 'prediction')
preds.show(10)

auc = evaluator.evaluate(preds)
auc

旁注：我使用的表格已经矢量化。该模型使用此代码运行，它运行缓慢（大约 10-15 分钟的训练时间）并且只使用 3-4 个（或者可能只使用其中一个）内核。

Answer 1

感谢上述 cmets 的澄清。

Spark 的实现不一定比 XGBoost 快。事实上，我会期待你所看到的。

最大的因素是 XGBoost 在设计和编写时特别考虑了梯度提升树。另一方面，Spark 的用途更广泛，而且很可能没有 XGBoost 所具有的那种优化。请参阅here，了解 XGBoost 和 scikit-learn 的分类器算法实现之间的区别。如果您想真正了解细节，可以阅读论文，甚至可以阅读 XGBoost 和 Spark 实现背后的代码。

请记住，XGBoost 也是并行/分布式的。它只是在同一台机器上使用多个线程。当数据不适合单台机器时，Spark 可帮助您运行算法。

我能想到的其他几个小问题是：a) Spark 确实有一个重要的启动时间。不同机器之间的通信也可以加起来。 b) XGBoost 是用 C++ 编写的，通常非常适合数值计算。

至于为什么 Spark 只使用 3-4 个核心，这取决于你的数据集大小，它是如何跨节点分布的，spark 启动的执行器数量是多少，占用了哪个阶段大多数时间，内存配置等。您可以使用 Spark UI 尝试找出发生了什么。如果不查看数据集，很难说出为什么会发生这种情况。

希望对您有所帮助。

编辑：我刚刚找到了一个比较简单 Spark 应用程序与独立 Java 应用程序之间的执行时间的好答案 - https://stackoverflow.com/a/49241051/5509005。同样的原则也适用于此，事实上，由于 XGBoost 已高度优化，因此更是如此。