如何在Spark中集成xgboost？（Python）

Question

我正在尝试使用XGBoost在Hive上的数据上训练模型。由于数据太大，我无法将其转换为Pandas DataFrame，因此必须使用Spark DataFrame。但是，在创建XGBoostEstimator时，出现了一个错误：

TypeError: 'JavaPackage' object is not callable
Exception AttributeError: "'NoneType' object has no attribute '_detach'" in ignored

我没有使用XGBoost for Spark的经验，尝试了一些在线教程，但都没有成功。我尝试将数据转换成Pandas DataFrame，但数据太大了，Java包装器总是提示OutOfMemoryException（我也试图查找解决方案，但对我来说都不起作用，即使增加执行程序内存也没用）。

我最近在跟随的教程是：

https://towardsdatascience.com/pyspark-and-xgboost-integration-tested-on-the-kaggle-titanic-dataset-4e75a568bdb

放弃使用XGBoost模块后，我开始使用sparkxgb。

spark = create_spark_session('shai', 'dna_pipeline')
# sparkxgboost files 
spark.sparkContext.addPyFile('resources/sparkxgb.zip')

def create_spark_session(username=None, app_name="pipeline"):
    if username is not None:
        os.environ['HADOOP_USER_NAME'] = username

    return SparkSession \
        .builder \
        .master("yarn") \
        .appName(app_name) \
        .config(...) \
        .config(...) \
        .getOrCreate()

def train():
    train_df = spark.table('dna.offline_features_train_full')
    test_df = spark.table('dna.offline_features_test_full')

    from sparkxgb import XGBoostEstimator

    vectorAssembler = VectorAssembler() \
        .setInputCols(train_df.columns) \
        .setOutputCol("features")

    # This is where the program fails
    xgboost = XGBoostEstimator(
        featuresCol="features",
        labelCol="label",
        predictionCol="prediction"
    )

    pipeline = Pipeline().setStages([xgboost])
    pipeline.fit(train_df)

完整的异常信息如下：

Traceback (most recent call last):
  File "/home/elad/DNA/dna/dna/run.py", line 283, in <module>
    main()
  File "/home/elad/DNA/dna/dna/run.py", line 247, in main
    offline_model = train_model(True, home_dir=config['home_dir'], hdfs_client=client)
  File "/home/elad/DNA/dna/dna/run.py", line 222, in train_model
    model = train(offline_mode=offline, spark=spark)
  File "/home/elad/DNA/dna/dna/model/xgboost_train.py", line 285, in train
    predictionCol="prediction"
  File "/home/elad/.conda/envs/DNAenv/lib/python2.7/site-packages/pyspark/__init__.py", line 105, in wrapper
    return func(self, **kwargs)
  File "/tmp/spark-7781039b-6821-42be-96e0-ca4005107318/userFiles-70b3d1de-a78c-4fac-b252-2f99a6761b32/sparkxgb.zip/sparkxgb/xgboost.py", line 115, in __init__
  File "/home/elad/.conda/envs/DNAenv/lib/python2.7/site-packages/pyspark/ml/wrapper.py", line 63, in _new_java_obj
    return java_obj(*java_args)
TypeError: 'JavaPackage' object is not callable
Exception AttributeError: "'NoneType' object has no attribute '_detach'" in <bound method XGBoostEstimator.__del__ of XGBoostEstimator_4f54b37156fb0a113233> ignored

我不知道为什么会出现这个异常，也不知道如何将SparkXGB正确地集成到我的代码中。

希望能得到帮助。

谢谢。

- Elad Cohen

Answer 1

在调试了整整一天的模块之后，问题只是提交jar包的方式不正确。我将jar包下载到本地，并使用以下命令进行pyspark-submit：

PYSPARK_SUBMIT_ARGS=--jars resources/xgboost4j-0.72.jar,resources/xgboost4j-spark-0.72.jar

这解决了问题。

- Elad Cohen

Answer 2

与其使用XGBoost，你可以尝试使用LightGBM。它是一种类似算法，且有人认为（至少速度更快）优于XGBoost。在pyspark中，它基本可以直接使用，你可以点击此处了解更多。

- Daniel

---

虽然我找到了解决方案，但我会查看这个模块，谢谢。 - Elad Cohen

Answer 3

新版本的Apache Spark(2.3.0)没有XGBoost。您应该尝试使用Pyspark。您必须将Spark数据帧转换为pandas数据帧。

这是一篇很棒的文章，提供了工作流程和解释xgboost and spark。

好的，我再次阅读了您的帖子，您声称数据集太大。也许您应该尝试Apache Arrow。请查看Speeding up Pyspark with Apache Arrow。

- Richard Rublev

---

你好，我尝试使用了许多工具，包括Arrow，但仍然无法解决问题。我得到了相同的“OutOfMemoryException”错误。 Spark本身没有XGBoost，但我正在尝试使用此API：https://github.com/dmlc/xgboost/pull/4656 - Elad Cohen

你的数据集实际大小是多少？ - Richard Rublev

大约6GB（更大的DFS大约为12GB） - Elad Cohen

尝试在某个地方上传一个dfs，然后发布链接。当然，如果数据不是机密的话。 - Richard Rublev

很遗憾，我无法分享数据。 - Elad Cohen