大数据是用于改善城市生活的 10 大领域之一。通过分析城市内的位置信息和行为模式，相关人员可以优化交通、作出更明智的规划决策，并提升广告的智能性。例如，通过分析 GPS 汽车数据，相关人员可以根据城市的实时交通信息优化交通流量。电信公司可利用手机定位数据，通过识别和预测大都市地区的位置活动趋势和人口模式来提供分析洞见。而且，现已证明，将机器学习 (ML) 应用于地理位置数据对于确定电信、旅游、市场营销和制造业的行业模式和发展趋势大有帮助。
在本章中，我们将使用纽约出租车行程公开数据来检验出租车行程数据的回归分析，因为该数据与预测纽约市出租车费用较为相关。首先，我们将概述 XGBoost 算法，然后探讨用例。
XGBoost 即极端梯度提升，是可扩展的分布式梯度提升决策树 (GBDT) 机器学习库。XGBoost 提供并行树提升功能，是应用于回归、分类和排序问题的出色 ML 库。RAPIDS 团队与 Distributed Machine Learning Common (DMLC) XGBoost 组织建立了紧密的合作关系，而且 XGBoost 现已加入无缝嵌入式 GPU 加速特性，可显著加快模型训练速度并提高准确性，从而得出更精确的预测结果。
梯度提升决策树 (GBDT) 是类似于随机森林的决策树集成算法，两者的区别在于树的构建和组合方式。随机森林使用名为 Bagging 的技术，通过数据集的自助抽样样本并行构建完整的决策树。最终得到的预测结果是所有决策树预测结果的平均值。梯度提升决策树使用名为 Boosting 的技术对一组浅层决策树进行迭代训练，每次迭代都会使用为前一个样本中的记录所赋予的权重（并未正确预测），从而减少后继树的错误。最终得到的预测结果是所有决策树预测结果的加权平均值。Bagging 可大幅减少差异和过拟合，而 Boosting 则可大幅减少偏差和欠拟合。
XGBoost 是 GBDT 的变体。使用 GBDT 时，决策树可按顺序构建。使用 XGBoost 时，决策树为并行构建，遵循 level-wise 生长策略，扫描梯度值并使用这些部分和来评估训练集中每个可分割点的分割质量。
GPU 加速的 XGBoost 算法利用并行前缀快速求和运算来扫描所有可能的分割，并通过并行基数排序对数据进行重新分区。此算法针对给定的提升迭代构建决策树，一次生成一层，并在 GPU 上同时处理整个数据集。
GPU 加速的 Spark XGBoost 提供以下关键特性：
示例数据集为纽约出租车数据集，该数据集已经过清理和转换，可添加特征，如使用此 Spark ETL notebook 的 Haversine 距离。
在此场景中，我们将基于以下特征构建模型，进而预测出租车费用：
首先，我们导入 Spark XGBoost 的 GPU 版本和 CPU 版本所需的软件包：
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
import org.apache.spark.sql._
import org.apache.spark.ml._
import org.apache.spark.ml.feature._
import org.apache.spark.ml.evaluation._
import org.apache.spark.sql.types._
import ml.dmlc.xgboost4j.scala.spark.{XGBoostRegressor, XGBoostRegressionModel}
使用 Spark XGBoost 的 GPU 版本时，您还需要进行以下导入操作：
import ml.dmlc.xgboost4j.scala.spark.rapids.{GpuDataReader, GpuDataset}
我们使用 Spark StructType 指定模式。
lazy val schema =
StructType(Array(
StructField("vendor_id", DoubleType),
StructField("passenger_count", DoubleType),
StructField("trip_distance", DoubleType),
StructField("pickup_longitude", DoubleType),
StructField("pickup_latitude", DoubleType),
StructField("rate_code", DoubleType),
StructField("store_and_fwd", DoubleType),
StructField("dropoff_longitude", DoubleType),
StructField("dropoff_latitude", DoubleType),
StructField(labelName, DoubleType),
StructField("hour", DoubleType),
StructField("year", IntegerType),
StructField("month", IntegerType),
StructField("day", DoubleType),
StructField("day_of_week", DoubleType),
StructField("is_weekend", DoubleType)
))
在以下代码中，我们创建 Spark 会话并设置训练和评估数据文件路径。（请注意：如果您使用 notebook，则不必创建 SparkSession。）
val trainPath = "/FileStore/tables/taxi_tsmall.csv"
val evalPath = "/FileStore/tables/taxi_esmall.csv"
val spark = SparkSession.builder().appName("Taxi-GPU").getOrCreate
我们将 CSV 文件的数据加载到 Spark DataFrame 中，并指定要加载到 DataFrame 中的数据源和模式，具体如下所示。
val tdf = spark.read.option("inferSchema",
"false").option("header", true).schema(schema).csv(trainPath)
val edf = spark.read.option("inferSchema", "false").option("header",
true).schema(schema).csv(evalPath)
DataFrame show(5) 显示前 5 行：
tdf.select("trip_distance", "rate_code","fare_amount").show(5)
result:
+------------------+-------------+-----------+
| trip_distance| rate_code|fare_amount|
+------------------+-------------+-----------+
| 2.72|-6.77418915E8| 11.5|
| 0.94|-6.77418915E8| 5.5|
| 3.63|-6.77418915E8| 13.0|
| 11.86|-6.77418915E8| 33.5|
| 3.03|-6.77418915E8| 11.0|
+------------------+-------------+-----------+
函数 Describe 返回一个 DataFrame，其中包含描述性汇总统计信息，例如计数、均值、标准差以及每个数字列的最小值和最大值。
tdf.select("trip_distance", "rate_code","fare_amount").describe().show
+-------+------------------+--------------------+------------------+
|summary| trip_distance| rate_code| fare_amount|
+-------+------------------+--------------------+------------------+
| count| 7999| 7999| 7999|
| mean| 3.278923615451919|-6.569284350812602E8|12.348543567945994|
| stddev|3.6320775770793547|1.6677419425906155E8|10.221929466939088|
| min| 0.0| -6.77418915E8| 2.5|
| max|35.970000000000006| 1.957796822E9| 107.5|
+-------+------------------+--------------------+------------------+
以下散点图用于探讨车费与行程距离之间的相关性。
%sql
select trip_distance, fare_amount
from taxi
需要将由 ML 算法使用的特征进行转换，并将其放入特征向量，这些向量是代表每个特征值的数字向量。如下所示，我们使用 VectorAssembler 转换器返回带有标签和向量特征列的新 DataFrame。
// 特征列名称
val featureNames = Array("passenger_count","trip_distance", "pickup_longitude","pickup_latitude","rate_code","dropoff_longitude", "dropoff_latitude", "hour", "day_of_week","is_weekend")
// 创建转换器
object Vectorize {
def apply(df: DataFrame, featureNames: Seq[String], labelName: String): DataFrame = {
val toFloat = df.schema.map(f => col(f.name).cast(FloatType))
new VectorAssembler()
.setInputCols(featureNames.toArray)
.setOutputCol("features")
.transform(df.select(toFloat:_*))
.select(col("features"), col(labelName))
}
}
// transform 方法添加特征列
var trainSet = Vectorize(tdf, featureNames, labelName)
var evalSet = Vectorize(edf, featureNames, labelName)
trainSet.take(1)
result:
res8: Array[org.apache.spark.sql.Row] =
Array([[5.0,2.7200000286102295,-73.94813537597656,40.82982635498047,-6.77418944E8,-73.96965026855469,40.79747009277344,10.0,6.0,1.0],11.5])
使用 XGBoost GPU 版本时，不需使用 VectorAssembler。
使用 CPU 版本时，应将 num_workers 设置为 CPU 内核数，将 tree_method 设置为“hist”，并在 Vector Assembler 中将特征列设置为输出特征列。
lazy val paramMap = Map(
"learning_rate" -> 0.05,
"max_depth" -> 8,
"subsample" -> 0.8,
"gamma" -> 1,
"num_round" -> 500
)
// 设置 xgboost 参数
val xgbParamFinal = paramMap ++ Map("tree_method" -> "hist", "num_workers" -> 12)
// 创建 xgboostregressor 估测器
val xgbRegressor = new XGBoostRegressor(xgbParamFinal)
.setLabelCol(labelName)
.setFeaturesCol("features")
使用 GPU 版本时，应将 num_workers 设置为 Spark 集群中具有 GPU 的计算机的数量，将 tree_method 设置为“gpu_hist”，并将特征列设置为包含特征名称的字符串数组。
val xgbParamFinal = paramMap ++ Map("tree_method" -> "gpu_hist",
"num_workers" -> 1)
// 创建估测器
val xgbRegressor = new XGBoostRegressor(xgbParamFinal)
.setLabelCol(labelName)
.setFeaturesCols(featureNames)
以下代码在训练数据集上使用 XGBoostRegressor 估测器拟合方法来训练并返回 XGBoostRegressor 模型。此外，我们还使用时间方法返回训练模型的时间，并使用此方法对采用 CPU 和 GPU 的训练时间进行比较。
object Benchmark {
def time[R](phase: String)(block: => R): (R, Float) = {
val t0 = System.currentTimeMillis
val result = block // call-by-name
val t1 = System.currentTimeMillis
println("Elapsed time [" + phase + "]: " +
((t1 - t0).toFloat / 1000) + "s")
(result, (t1 - t0).toFloat / 1000)
}
}
// 使用估测器来拟合（训练）模型
val (model, _) = Benchmark.time("train") {
xgbRegressor.fit(trainSet)
}
尚未用于训练的评估数据集也可用于评估模型的性能。我们使用模型转换方法对测试数据进行预测。
此模型将使用经训练的 XGBoost 模型进行估测，然后在返回的 DataFrame 的新预测列中返回车费预测结果。此处，我们再次使用基准测试时间方法来比较预测时间。
val (prediction, _) = Benchmark.time("transform") {
val ret = model.transform(evalSet).cache()
ret.foreachPartition(_ => ())
ret
}
prediction.select( labelName, "prediction").show(10)
Result:
+-----------+------------------+
|fare_amount| prediction|
+-----------+------------------+
| 5.0| 4.749197959899902|
| 34.0|38.651187896728516|
| 10.0|11.101678848266602|
| 16.5| 17.23284912109375|
| 7.0| 8.149757385253906|
| 7.5|7.5153608322143555|
| 5.5| 7.248467922210693|
| 2.5|12.289423942565918|
| 9.5|10.893491744995117|
| 12.0| 12.06682014465332|
+-----------+------------------+
RegressionEvaluator 评估方法将计算预测列和标签列的均方根误差，即均方误差的平方根。
val evaluator = new RegressionEvaluator().setLabelCol(labelName)
val (rmse, _) = Benchmark.time("evaluation") {
evaluator.evaluate(prediction)
}
println(s"RMSE == $rmse")
Result:
Elapsed time [evaluation]: 0.356s
RMSE == 2.6105287283128353
如下所示，可将模型保存到磁盘，以便以后使用。
model.write.overwrite().save(savepath)
保存模型可得到一个元数据的 JSON 文件和一个模型数据的 Parquet 文件。我们可以使用加载命令重新加载模型。原始模型和重新加载的模型相同。
val sameModel = XGBoostRegressionModel.load(savepath)
在本章中，我们介绍了 XGBoost 的工作原理，以及如何将 XGBoost 回归和 Spark 结合使用来预测出租车费用。现在，您可以利用大型数据集，在 CPU 和 GPU 上运行此示例，以比较预测的时间和准确性。
