巨量資料是都市改善的十大領域之一。分析都市內的地點和行為模式，可最佳化交通運輸、做出更完善的規劃決策，並提供更具智慧性的廣告。例如，對 GPS 汽車資料的分析讓都市能夠根據即時交通資訊，使交通運輸流量最佳化。電信公司利用行動電話定位資料，透過識別和預測大都會地區人口的地點活動趨勢和模式，提供相關見解。除此之外，將機器學習 (ML) 應用於地理定位資料，對於識別 電信、旅遊、行銷和製造業 中的模式和趨勢上也有所助益。
在這一章節中，我們會使用紐約的公共計程車車程資料來檢查其迴歸分析，因為此資料和預測紐約市計程車車資有所關聯。我們會從 XGBoost 演算法的概述開始，然後探討使用案例。
XGBoost，代表極端梯度提升 (Extreme Gradient Boosting)，是種可擴充的分散式梯度提升決策樹 (GBDT) 機器學習函式庫。XGBoost 提供的平行決策樹提升，是處理迴歸、分類和排序問題的主要機器學習函式庫。RAPIDS 團隊與 Distributed Machine Learning Common (DMLC) XGBoost 組織密切合作，現在 XGBoost 已加入了流暢的拖曳式 (drop-in) GPU 加速，能顯著加速模型訓練並提高預測準確性。
梯度提升決策樹 (GBDT) 是種類似隨機森林的決策樹集成演算法，區別在於決策樹的建造和組合方式。隨機森林使用一種稱為 Bagging 的技術，從資料集的自助樣本中平行建立完整的決策樹。最終預測是所有決策樹預測的平均值。梯度提升決策樹使用一種稱為提升的技術，來反覆訓練一個淺度決策樹的集成，每次迭代時對於在前一個樣本中未正確預測的紀錄均給予權重，透過這個方式來減少後續決策樹的錯誤。最終預測是所有決策樹預測的加權平均值。Bagging 可將差異和過度擬合最小化，而提升可將偏差和欠擬合 (underfitting) 最小化。
XGBoost 是 GBDT 的一個變異。透過 GBDT，決策樹是按順序建立的。透過 XGBoost，決策樹是平行建立的，遵循同一層無差別 (level-wise) 的策略進行跨梯度值掃描，並使用這些部分加總來評估訓練集內每個可能分段 (split) 的分段品質。
GPU 加速的 XGBoost 演算法使用快速平行前置加總運算來掃描所有可能的分段，以及平行基數排序 (parallel radix sorting) 以重新分割資料。這種演算法為指定的提升反覆迭代建立決策樹，一次建立一個層級，同時在 GPU 上處理整個資料集。
GPU 加速的 XGBoost 提供以下主要功能：
範例資料集是紐約計程車資料集，並已經過清理和轉換來新增特徵，例如使用此 Spark ETL 筆記型電腦計算的半正矢距離 (Haversine Distance)。
在這個案例中，我們會根據以下特徵建立模型來預測計程車車資金額：
首先，我們匯入 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 工作階段並設定訓練和評估資料的檔案路徑。(注意：如果您使用的是筆記型電腦，則不需要建立 Spark 工作階段。)
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
要使機器學習演算法使用特徵，必須將其轉換並放入特徵向量中，也就是代表每個特徵值的數字向量。以下內容使用向量合併 (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))
}
}
// 轉換方法加入特徵欄
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 版本時，不需要向量合併器。
在 CPU 版本中，num_workers 應設定為 CPU 核心的數量、tree_method 設為「hist」，及將特徵欄設為在向量合併器中的輸出特徵欄。
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 使用 XGBoost 迴歸預測計程車車資金額。現在，您可以在 CPU 和 GPU 上執行此範例，運用更大的資料集來比較預測的時間和準確性。