XGBoost4J-Spark-GPU 教程 (版本 1.6.1+)

XGBoost4J-Spark-GPU 是一个开源库，旨在通过利用 RAPIDS Accelerator for Apache Spark 产品，在 Apache Spark 集群上使用 GPU 从端到端加速分布式 XGBoost 训练。

本教程将向您展示如何使用 XGBoost4J-Spark-GPU。

使用 XGBoost4J-Spark-GPU 构建 ML 应用程序

将 XGBoost 添加到您的项目中

在深入学习如何使用 XGBoost4J-Spark-GPU 的教程之前，建议参考从 Maven 仓库安装以获取将 XGBoost4J-Spark-GPU 添加为项目依赖的说明。我们为您提供了稳定版本和快照版本，以供选择。

数据准备

在本节中，我们使用 Iris 数据集作为示例，展示如何使用 Apache Spark 转换原始数据集，使其适应 XGBoost 的数据接口。

Iris 数据集以 CSV 格式提供。每个实例包含 4 个特征，即“萼片长度”、“萼片宽度”、“花瓣长度”和“花瓣宽度”。此外，它还包含“类别”列，该列本质上是一个标签，具有三个可能的值：“Iris Setosa”、“Iris Versicolour”和“Iris Virginica”。

使用 Spark 内置的读取器读取数据集

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DoubleType, StringType, StructField, StructType}

val spark = SparkSession.builder().getOrCreate()

val labelName = "class"
val schema = new StructType(Array(
    StructField("sepal length", DoubleType, true),
    StructField("sepal width", DoubleType, true),
    StructField("petal length", DoubleType, true),
    StructField("petal width", DoubleType, true),
    StructField(labelName, StringType, true)))

val xgbInput = spark.read.option("header", "false")
    .schema(schema)
    .csv(dataPath)

首先，我们创建一个 SparkSession 实例，这是任何使用 DataFrame 的 Spark 应用程序的入口点。schema 变量定义了封装 Iris 数据的 DataFrame 的结构。通过显式设置的结构，我们可以定义列名及其类型；否则，列名将是 Spark 默认生成的，如 _col0 等。最后，我们可以使用 Spark 内置的 CSV 读取器将 Iris CSV 文件加载为名为 xgbInput 的 DataFrame。

Apache Spark 还包含许多用于其他格式（如 ORC、Parquet、Avro、JSON）的内置读取器。

转换原始鸢尾花数据集

为了让XGBoost识别Iris数据集，我们需要将字符串类型的标签，即“类别”，编码为双精度类型的标签。

将字符串类型的标签转换为双精度类型的一种方法是使用 Spark 的内置特征转换器 StringIndexer。但这个功能在 RAPIDS Accelerator 中没有加速，这意味着它将回退到 CPU。相反，我们使用以下代码通过另一种方式实现相同的目标：

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

val spec = Window.orderBy(labelName)
val Array(train, test) = xgbInput
    .withColumn("tmpClassName", dense_rank().over(spec) - 1)
    .drop(labelName)
    .withColumnRenamed("tmpClassName", labelName)
    .randomSplit(Array(0.7, 0.3), seed = 1)

train.show(5)

+------------+-----------+------------+-----------+-----+
|sepal length|sepal width|petal length|petal width|class|
+------------+-----------+------------+-----------+-----+
|         4.3|        3.0|         1.1|        0.1|    0|
|         4.4|        2.9|         1.4|        0.2|    0|
|         4.4|        3.0|         1.3|        0.2|    0|
|         4.4|        3.2|         1.3|        0.2|    0|
|         4.6|        3.2|         1.4|        0.2|    0|
+------------+-----------+------------+-----------+-----+

通过窗口操作，我们将标签的字符串列映射到标签索引。

训练

XGBoost-Spark 的 GPU 版本支持回归和分类模型。尽管我们在本教程中使用 Iris 数据集来展示如何使用 XGBoost/XGBoost4J-Spark-GPU 解决多类分类问题，但回归中的使用方法与分类非常相似。

要训练一个用于分类的 XGBoost 模型，我们首先需要定义一个 XGBoostClassifier：

import ml.dmlc.xgboost4j.scala.spark.XGBoostClassifier
val xgbParam = Map(
    "objective" -> "multi:softprob",
    "num_class" -> 3,
    "num_round" -> 100,
    "device" -> "cuda",
    "num_workers" -> 1)

val featuresNames = schema.fieldNames.filter(name => name != labelName)

val xgbClassifier = new XGBoostClassifier(xgbParam)
    .setFeaturesCol(featuresNames)
    .setLabelCol(labelName)

device 参数用于通知 XGBoost 应使用 CUDA 设备而不是 CPU。与单节点模式不同，GPU 由 Spark 管理，而不是由 XGBoost 管理。因此，不支持像 cuda:1 这样显式指定的设备序号。

训练 XGBoost 模型的可用参数可以在这里找到。与 XGBoost4J-Spark 包类似，除了默认参数集外，XGBoost4J-Spark-GPU 还支持这些参数的驼峰命名变体，以与 Spark 的 MLlib 命名约定保持一致。

具体来说，此页面中的每个参数在 XGBoost4J-Spark-GPU 中都有其对应的驼峰命名形式。例如，要为每棵树设置 max_depth，你可以像我们在上面的代码片段中那样传递参数（作为 max_depth 包装在 Map 中），或者你可以通过 XGBoostClassifer 中的设置器来实现：

val xgbClassifier = new XGBoostClassifier(xgbParam)
    .setFeaturesCol(featuresNames)
    .setLabelCol(labelName)
xgbClassifier.setMaxDepth(2)

备注

与接受VectorUDT类型特征列和特征列名称数组的XGBoost4j-Spark不同，XGBoost4j-Spark-GPU仅通过``setFeaturesCol(value: Array[String])``接受特征列名称数组。

在设置 XGBoostClassifier 参数和特征/标签列之后，我们可以通过使用输入 DataFrame 拟合 XGBoostClassifier 来构建一个转换器，即 XGBoostClassificationModel。这个 fit 操作本质上就是训练过程，生成的模型随后可以用于其他任务，如预测。

val xgbClassificationModel = xgbClassifier.fit(train)

预测

当我们获取一个模型时，无论是 XGBoostClassificationModel 还是 XGBoostRegressionModel，它都会以 DataFrame 作为输入，读取包含特征向量的列，为每个特征向量进行预测，并默认输出一个包含以下列的新 DataFrame：

XGBoostClassificationModel 将输出边缘 (rawPredictionCol)、概率 (probabilityCol) 以及每个可能标签的最终预测标签 (predictionCol)。
XGBoostRegressionModel 将输出一个预测标签（predictionCol）。

val xgbClassificationModel = xgbClassifier.fit(train)
val results = xgbClassificationModel.transform(test)
results.show()

通过上述代码片段，我们得到一个包含每个实例的边际、每个类别的概率以及每个实例的预测的DataFrame。

+------------+-----------+------------------+-------------------+-----+--------------------+--------------------+----------+
|sepal length|sepal width|      petal length|        petal width|class|       rawPrediction|         probability|prediction|
+------------+-----------+------------------+-------------------+-----+--------------------+--------------------+----------+
|         4.5|        2.3|               1.3|0.30000000000000004|    0|[3.16666603088378...|[0.98853939771652...|       0.0|
|         4.6|        3.1|               1.5|                0.2|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         4.8|        3.1|               1.6|                0.2|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         4.8|        3.4|               1.6|                0.2|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         4.8|        3.4|1.9000000000000001|                0.2|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         4.9|        2.4|               3.3|                1.0|    1|[-2.1498908996582...|[0.00596602633595...|       1.0|
|         4.9|        2.5|               4.5|                1.7|    2|[-2.1498908996582...|[0.00596602633595...|       1.0|
|         5.0|        3.5|               1.3|0.30000000000000004|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         5.1|        2.5|               3.0|                1.1|    1|[3.16666603088378...|[0.98853939771652...|       0.0|
|         5.1|        3.3|               1.7|                0.5|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         5.1|        3.5|               1.4|                0.2|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         5.1|        3.8|               1.6|                0.2|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         5.2|        3.4|               1.4|                0.2|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         5.2|        3.5|               1.5|                0.2|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         5.2|        4.1|               1.5|                0.1|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         5.4|        3.9|               1.7|                0.4|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         5.5|        2.4|               3.8|                1.1|    1|[-2.1498908996582...|[0.00596602633595...|       1.0|
|         5.5|        4.2|               1.4|                0.2|    0|[3.25857257843017...|[0.98969423770904...|       0.0|
|         5.7|        2.5|               5.0|                2.0|    2|[-2.1498908996582...|[0.00280966912396...|       2.0|
|         5.7|        3.0|               4.2|                1.2|    1|[-2.1498908996582...|[0.00643939292058...|       1.0|
+------------+-----------+------------------+-------------------+-----+--------------------+--------------------+----------+

提交申请

假设你已经配置了支持GPU的Spark独立集群。否则，请参考 spark standalone configuration with GPU support。

从 XGBoost 2.1.0 开始，阶段级调度自动启用。因此，如果您使用的是 Spark 独立集群版本 3.4.0 或更高版本，我们强烈建议将 "spark.task.resource.gpu.amount" 配置为分数值。这将允许在 ETL 阶段并行运行多个任务。一个示例配置是 "spark.task.resource.gpu.amount=1/spark.executor.cores"。然而，如果您使用的是早于 2.1.0 的 XGBoost 版本或低于 3.4.0 的 Spark 独立集群版本，您仍然需要将 "spark.task.resource.gpu.amount" 设置为等于 "spark.executor.resource.gpu.amount"。

备注

目前，XGBoost 中的阶段级调度功能仅限于 Spark 独立集群模式。然而，我们计划在 Spark 3.5.1 正式发布后，将其兼容性扩展到 YARN 和 Kubernetes。

假设应用程序的主类是“Iris”，应用程序的jar包是“iris-1.0.0.jar”，下面是一个实例，演示如何将xgboost应用程序提交到Apache Spark Standalone集群。

rapids_version=23.10.0
xgboost_version=2.0.1
main_class=Iris
app_jar=iris-1.0.0.jar

spark-submit \
  --master $master \
  --packages com.nvidia:rapids-4-spark_2.12:${rapids_version},ml.dmlc:xgboost4j-gpu_2.12:${xgboost_version},ml.dmlc:xgboost4j-spark-gpu_2.12:${xgboost_version} \
  --conf spark.executor.cores=12 \
  --conf spark.task.cpus=1 \
  --conf spark.executor.resource.gpu.amount=1 \
  --conf spark.task.resource.gpu.amount=0.08 \
  --conf spark.rapids.sql.csv.read.double.enabled=true \
  --conf spark.rapids.sql.hasNans=false \
  --conf spark.plugins=com.nvidia.spark.SQLPlugin \
  --class ${main_class} \
   ${app_jar}

首先，我们需要通过 --packages 指定 RAPIDS Accelerator, xgboost4j-gpu, xgboost4j-spark-gpu 包
其次，RAPIDS Accelerator 是一个 Spark 插件，因此我们需要通过指定 spark.plugins=com.nvidia.spark.SQLPlugin 来配置它。

有关其他 RAPIDS Accelerator 配置的详细信息，请参阅配置。

关于 RAPIDS Accelerator 常见问题，请参阅常见问题。