SparkMLlib GBDT算法工业大数据实战

在格物汇之前发表的《工业大数据挖掘的利器——Spark MLlib》中提到，Spark 的MLlib组件能够对工业现场海量数据进行高效挖掘，快速呈现结果给业务分析人员。接下来将向大家介绍SparkMLlib 中的GBDT算法，并将应用该算法对工业数据进行代码实战。

1算法概念

GB（Gradient Boosting）梯度提升算法，GB 共需要进行M次迭代，通过采用梯度下降的方法，每次迭代向损失函数的负梯度方向进行移动，从而使损失函数越来越小，进而使模型越来越精确。算法伪代码如下：

GB算法跟原始的Boosting算法相比较，还是有比较明显的区别。

Boosting算法开始的时候，是会给每个样本附上权重的，在每次迭代的时候就会增加错的样本的权重，减少对的样本的权重，经过N次迭代之后，会得到N个分类器，然后我们再将他们组合起来，得到最终模型。

GB算法与Boosting区别是，他的每一次迭代的目标都是减少上一次的残差，所以在残差减少的方向上建立一个新的模型。在GB算法框架上加入决策树，就是GBDT（GradientBoost Decision Tree）算法。

GBDT主要的优点有：

1) 可以灵活处理各种类型的数据，包括连续值和离散值。

2) 在相对少的调参时间情况下，预测的准备率也可以比较高。这个是相对SVM来说的。

3)使用一些健壮的损失函数，对异常值的鲁棒性非常强。比如 Huber损失函数和Quantile损失函数。

4) 很好的利用了弱分类器进行级联。

5) 充分考虑的每个分类器的权重。

6) 可以得到变量间的重要性排序。

GBDT的主要缺点有：

1）由于弱学习器之间存在依赖关系，难以并行训练数据，不过可以通过自采样的SGBT来达到部分并行。

1完整代码实例

工业生产中，产品在制程过程中会有很多特性值，如果能对产品的特性值及时进行预测，得到特性值的具体数值，那么就会帮组业务人员知晓产品的质量，实现产品的全检，并能防止异常产品后流，造成不必要的浪费。

本次实战代码的采用的数据是半导体制程中某一道工序的机台的制程参数值，通过采用SparkMLlib中的GBDT算法对工业现场机台的制程参数进行建模，预测出经过该机台生产之后产品的膜层厚度。

packageSparkML

importcommon.Logger
importorg.apache.spark.ml.Pipeline
importorg.apache.spark.ml.evaluation.{BinaryClassificationEvaluator,RegressionEvaluator}
importorg.apache.spark.ml.feature.VectorAssembler
importorg.apache.spark.ml.regression.GBTRegressor
importorg.apache.spark.ml.tuning.{CrossValidator,ParamGridBuilder}
importorg.apache.spark.sql.{Row,SparkSession}
importscala.collection.mutable.ArrayBuffer
/**
* Created by huanghuan01 on 2019/3/27.
*/
objectgbdtDemoextendsLogger{

defmain(args: Array[String]):Unit= {
valspark= SparkSession
.builder()
.enableHiveSupport()
.master("local[4]")
.appName("gbdtDemo")
.getOrCreate()

spark.sparkContext.setLogLevel("WARN")

varrawData= spark.read.format("csv")
.option("header","true")
.load("E:\\sampleData.csv")

valfieldNames= rawData.schema.map(f=>s"${f.name}").toArray

valcastBuffer:ArrayBuffer[String] = ArrayBuffer()
for(i<-0until fieldNames.length){
valcast_str="cast("+ fieldNames(i) +" as double) as "+ fieldNames(i)
castBuffer.append(cast_str)
}
valcastArr= castBuffer.toArray
valinputData = rawData.selectExpr(castArr:_*)
valfeatureFieldNames= fieldNames.filter(!_.contains("label"))

valfeatureIndexer=newVectorAssembler()
.setInputCols(featureFieldNames)
.setOutputCol("featureIndexer")

valgbt=newGBTRegressor()
.setLabelCol("label")
.setFeaturesCol("featureIndexer")

valArray(trainingData,testData) =inputData.randomSplit(Array(0.8,0.2))

valpipline =newPipeline()
.setStages(Array(featureIndexer,gbt))

valparamGrid =newParamGridBuilder()

.addGrid(gbt.maxIter,Array(30,50,100,200))

.addGrid(gbt.maxDepth,Array(3,7,9))

.addGrid(gbt.stepSize,Array(0.01,0.05,0.1))

.build()

valcv =newCrossValidator()
.setEstimator(pipline)
.setEvaluator(newRegressionEvaluator())
.setNumFolds(5)
.setEstimatorParamMaps(paramGrid)
valmodel =cv.fit(trainingData)

valpredictions =model.transform(testData)

predictions.select("label","prediction").show(100,false)

valevaluator =newRegressionEvaluator()
.setLabelCol("label")
.setPredictionCol("prediction")
.setMetricName("mae")

val mae = evaluator.evaluate

(predictions)
log.warn(s"The mae is : ${mae}")

val predictionAndLabels =

predictions

.select("prediction",

"label")

.rdd

.map { case Row(prediction:

Double, label: Double) =>

(prediction, label) }

val mape = math.abs

(predictionAndLabels.map

{ x => math.abs((x._1 - x._2) /

x._1) }.mean())

log.warn(s"The mape is :

${mape}")

val pipLine = model.bestModel.

asInstanceOf[org.apache.spark.

ml.PipelineModel]

}
}

模型最后输出模型性能指标如下：

Mape(Mean Absolute Percentage Error):0.23%

通过上图模型输出的预测值与实际值对比，发现预测出来的产品膜厚的数值走势跟实际数值走势基本符合，mape达到0.5%以内，拟合度相当可观，后续还可以通过样本筛选以及特征工程等手段对该模型进行进一步调优。

在模型达到业务需求的拟合度等指标后，通过该模型进行部署，实现产品的“实时全检”，从而实现产品质量的全面监控，杜绝异常产品后流；与工厂内的抽检系统结合后，降低产品的抽检率，提高工厂的效率。

GBDT算法的用途还是比较广泛的，它不仅可以处理分类问题，能对线性与非线性回归问题进行处理，还能通过输出变量间重要因子排序，方便业务人员快速定位异常变量。在工业现场的顽固异常分析还是产品特性预测等领域，GBDT算法确实是很值得数据分析人员考虑的一种算法。

本文作者：

格创东智大数据工程师黄欢（转载请注明作者及来源）