技术好文：Spark机器学习笔记一

Spark机器学习库现支持两种接口的API:RDD-based和DataFrame-based，Spark官方网站上说，RDD-based APIs在2.0后进入维护//代码效果参考：http://hnjlyzjd.com/xl/wz_25104.html

在学习了两周Spark MLlib后，准备转向DataFrame-based接口。由于现有的文档资料均是RDD-based接口，于是便去看了看Spark MLlib的源码。DataFrame-based API 包含在org.apache.spark.ml包中，其中主要的类结构如下：

咱先看一个线性回归的例子examples/ml/LinearRegressionExample.scala，其首先定义了一个LinearRegression的对象:

val lir = new LinearRegression()

.setFeaturesCol("features")

.setLabelCol("label")

.setRegParam(params.regParam)

.setElasticNetParam(params.elasticNetParam)

.setMaxIter(params.maxIter)

.setTol(params.tol)

然后，调用fit方法训练数据，得到一个训练好的模型lirModel，它是一个LinearRegressionModel类的对象。

val lirModel = lir.fit(training)

现在，我们大概可以理清MLlib机器学习的流程，和很多单机机器学习库一样，先定义一个模型并设置好参数，然后训练数据，最后返回一个训练好了的模型。

我们现在在源码中去查看LinearRegression和LinearRegressionModel，其类的依赖关系如下：

LinearRegression是一个Predictor，LinearRegressionModel是一个Model，那么Predictor是学习算法，Model是训练得到的模型。除此之外，还有一类继承自Params的类，这是一个表示参数的类。Predictor 和Model 共享一套参数。

现在用Spark MLlib来完成第一个机器学习例子，数据是我之前放在txt文件里的回归数据，一共550多万条，共13列，第一列是Label，后面是Features。分别演示两种接口，先用旧的接口：

1.读取原始数据：

scala> import org.apache.spark.mllib.linalg.

import org.apache.spark.mllib.linalg.

scala

import org.apache.spark.mllib.regression._

scala

raw_data: org.apache.spark.rdd.RDD【String】 = data/my/y_x.txt MapPartitionsRDD【1】 at textFile at :30

2.转换格式，RDD-based接口以LabeledPoint为输入数据的格式：

scala> val data = rawdata.map{ line =>

| val arr = line.split(' ').map(.toDouble)

| val label = arr.head

| val features = Vectors.dense(arr.tail)| LabeledPoint(label,features)

| }

data: org.apache.spark.rdd.RDD【org.apache.spark.mllib.regression.LabeledPoint】 = MapPartitionsRDD【2】 at map at :32

3.划分train、test数据集：

scala> val splits = data.randomSplit(Array(0.8, 0.2))

splits: Array【org.apache.spark.rdd.RDD【org.apache.spark.mllib.regression.LabeledPoint】】 = Array(MapPartitionsRDD【3】 at randomSplit at :34, MapPartitionsRDD【4】 at randomSplit at :34)

scala

train_set: org.apache.spark.rdd.RDD【org.apache.spark.mllib.regression.LabeledPoint】 = MapPartitionsRDD【3】 at randomSplit at :34

scala

test_set: org.apache.spark.rdd.RDD【org.apache.spark.mllib.regression.LabeledPoint】 = MapPartitionsRDD【4】 at randomSplit at :34

4.使用LinearRegressionWithSGD.train训练模型：

scala> val lr = LinearRegressionWithSGD.train(train_set,100,0.0001)

warning: there was one deprecation warning; re-run with -deprecation for details

16/08/26 09:20:44 WARN Executor: 1 block locks were not released by TID = 0:

【rdd_3_0】

lr: org.apache.spark.mllib.regression.LinearRegressionModel = org.apache.spark.mllib.regression.LinearRegressionModel: intercept = 0.0, numFeatures = 12

5.模型评估：

scala> val pred_labels = test_set.map(lp => (lp.label, lr.predict(lp.features)))

predlabels: org.apache.spark.rdd.RDD【(Double, Double)】 = MapPartitionsRDD【17】 at map at :42

scala math.pow(p-v,2)}.mean

mse: Double = 0.05104150735910074

再用新的接口：

1.读取原始数据：

scala> import org.apache.spark.ml.linalg.

import org.apache.spark.ml.linalg.

scala</span] import org.apache.spark.ml.regression.

import org.apache.spark.ml.regression.

scala</span] import org.apache.spark.sql.

import org.apache.spark.sql._

scala

raw_data: org.apache.spark.sql.DataFrame = 【value: string】

2.转换数据

scala> val data = rawdata.rdd.map { case Row(line:String) =>

| val arr = line.split(' ').map(.toDouble)

| val label = arr.head

| val features = Vectors.dense(arr.tail)

| (label,features)

| }

data: org.apache.spark.rdd.RDD【(Double, org.apache.spark.ml.linalg.Vector)】 = MapPartitionsRDD【4】 at map at :34

3.划分数据集

scala> val splits = data.randomSplit(Array(0.8, 0.2))

splits: Array【org.apache.spark.rdd.RDD【(Double, org.apache.spark.ml.linalg.Vector)】】 = Array(MapPartitionsRDD【5】 at randomSplit at :36, MapPartitionsRDD【6】 at randomSplit at :36)

scala

train_set: org.apache.spark.sql.Dataset【(Double, org.apache.spark.ml.linalg.Vector)】 = 【_1: double, _2: vector】

scala

test_set: org.apache.spark.sql.Dataset【(Double, org.apache.spark.ml.linalg.Vector)】 = 【_1: double, _2: vector】

4.创建LinearRegression对象，并设置模型参数。这里设置类LabelCol和FeaturesCol列，默认为“label”和“features”，而我们的数据是"_1"和”_2“。

scala> val lir = new LinearRegression

lir: org.apache.spark.ml.regression.LinearRegression = linReg_c4e70a01bcd3

scala

res0: org.apache.spark.ml.regression.LinearRegression = linReg_c4e70a01bcd3

scala

res1: org.apache.spark.ml.regression.LinearRegression = linReg_c4e70a01bcd3

5.训练模型

val model = lir.fit(train_set)

16/08/26 09:45:16 WARN Executor: 1 block locks were not released by TID = 0:

【rdd_9_0】

16/08/26 09:45:16 WARN WeightedLeastSquares: regParam is zero, which might cause numerical instability and overfitting.

model: org.apache.spark.ml.regression.LinearRegressionModel = linReg_c4e70a01bcd3

6.模型评估

scala> val res = model.transform(test_set)

res: org.apache.spark.sql.DataFrame = 【_1: double, 2: vector ... 1 more field】

scala</span] import org.apache.spark.ml.evaluation.

import //代码效果参考：http://hnjlyzjd.com/xl/wz_25102.html

scala

eva: org.apache.spark.ml.evaluation.RegressionEvaluator = regEval_8fc6cce63aa9

scala

res6: eva.type = regEval_8fc6cce63aa9

scala

res7: eva.type = regEval_8fc6cce63aa9

scala

res8: Double = 0.027933653533088666