大数据开发之机器学习总结（Spark Mllib）（四）

背景

1. 在大数据和机器学习交叉的领域，如果公司选择了hadoop生态，结合spark框架，则spark 的mllib用于机器学习实际应用就是不二选择了。
2. 团队有spark基础，学习和适用门槛低。但如果选择python生态，则需要团队有python基础，另外个人认为，python工程化对比java生态还是差了那么一些意思。

1. Spark MLLib简介

1. spark的mllib目前支持4种常见机器学习问题，分类，回归，聚类，协同过滤。

2. 

3. mllib本身还是基于RDD算子实现，所以天生就可以和spark sql，spark streaming无缝结成，也就是spark的四个模块spark sql，spark streaming， spark mllib， graphX可以直接解决大数据数据计算的90%以上场景，覆盖结构化数据处理，流式数据处理，机器学习，图计算。
4. mllib的架构图

   

3个组成部分

底层基础：包括Spark的运行库、矩阵库和向量库；

算法库：包含广义线性模型、推荐系统、聚类、决策树和评估的算法；

实用程序：包括测试数据的生成、外部数据的读入等功能。

1. mllib底层基础

• 基础部分主要包括向量接口Vector和矩阵接口Matrix，这两种接口都会使用Scala语言基于Netlib和BLAS/LAPACK开发的线性代数库Breeze。

这里结合另外一篇数学基础博客，可以直到如果要处理分类，聚类，协同过滤，很多时候就是将事物特征向量提取出来，然后适用算法对向量做相似度或者空间距离计算

而矩阵可以看成是向量集合，更方便向量的计算，因为是一批向量组合为一个矩阵。

• MLlib支持本地的密集向量和稀疏向量，并且支持标量向量。

  注意，在海量数据处理时，如果适用类似数组等集合表示向量，但如果一个10000长度的数组中大部分数据都是相同的如0，少量是1，则没必要真的创建10000长度数组，直接以坐标数组和对应坐标值数组表示会更加节省空间。

  如Array(1,8,9,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,18,0,0,0,0,10,0,0,0,0,0,0,0,0,0,11,0,0,0,0,0,89,0,0,0,0,0);这时候就可以采取比这种更加节省内存方式，如tuple: ( 1000, Array[0,1,2,54,63,86,90], Array[1,8,9,18,10,11,89] )，就可以表示一个1000长度数组，index对应是Array[0,1,2,54,63,86,90]，对应值是Array[1,8,9,18,10,11,89] 。其他地方都是0，这种叫做稀疏性向量

  类似[1,8,9,76,1,23,78,66,12,35,0,0,0,65,87,8,276,28,8,88,98]，这种称之为密集型向量

spark中，稀疏和密集型向量对象就是：DenseVector 和SparseVector

示例

稀疏和密集向量在计算和存储时数据对比

疏矩阵在含有大量非零元素的向量Vector计算中会节省大量的空间并大幅度提高计算速度，如下图所示。

标量LabledPoint在实际中也被大量使用，例如判断邮件是否为垃圾邮件时就可以使用类似于以下的代码：

可以把表示为1.0的判断为正常邮件，而表示为0.0则作为垃圾邮件来看待。

对于矩阵Matrix而言，本地模式的矩阵如下所示。

• MLlib同时支持本地矩阵和分布式矩阵，支持的分布式矩阵分为RowMatrix、IndexedRowMatrix、CoordinateMatrix等。

在大数据处理时，分布式矩阵的引入更有意义，当然也支持本地矩阵。

分布式计算最主要就是数据切分和任务切分，这样可以充分利用各个节点的并行计算能力。其他考虑点则是优化数据切分策略，数据传输策略，中间计算结果保存，计算结果保存，任务切分，任务调度等等因素。

1. 矩阵编程接口

• RowMatrix直接通过RDD[Vector]来定义并可以用来统计平均数、方差、协同方差等：

  

  

平均数，如何计算就不赘述

均值描述的是样本集合的中间点，它告诉我们的信息是很有限的

标准差

标准差给我们描述的则是样本集合的各个样本点到均值的距离之平均

方差

方差则仅仅是标准差的平方

协方差

• IndexedRowMatrix是带有索引的Matrix，但其可以通过toRowMatrix方法来转换为RowMatrix，从而利用其统计功能，代码示例如下所示。

  

• CoordinateMatrix常用于稀疏性比较高的矩阵，是由RDD[MatrixEntry]来构建的，MatrixEntry是一个Tuple类型的元素，其中包含行、列和元素值，代码示例如下所示：

  

1. 特征向量处理（规范化）

   把原始数据中的值直接转入向量，各特征的值量纲可能差距很大，对算法的效果产生巨大负面效应（某些特征可能会掩盖掉其他特征的作用）所以需要对向量进行缩放（规范化、归一化）,mllib中有四种工具

• 范数规范器：Normalizer

  针对一行来操作！

  .setP(2) 2阶P范数

  

import org.apache.spark.ml.feature.Normalizer

// 正则化每个向量到1阶范数
val normalizer = new Normalizer()
  .setInputCol("features")
  .setOutputCol("normFeatures")
  .setP(1.0)

val l1NormData = normalizer.transform(dataFrame)
println("Normalized using L^1 norm")
l1NormData.show()

// 将每一行的规整为1阶范数为1的向量，1阶范数即所有值绝对值之和。
+-----+---------------------+-----------------------------------+
| id  |   features   |      normFeatures   |
+-----+----------------------+----------------------------------+
|  0| [1.0,0.5,-1.0]  |    [0.4,0.2,-0.4]       |
|  1| [2.0,1.0,1.0]   |   [0.5,0.25,0.25]      |
|  2| [4.0,10.0,2.0]  |[0.25,0.625,0.125]      |
+---+------------------------+---------------------------------+

// 正则化每个向量到无穷阶范数
val lInfNormData = normalizer.transform(dataFrame, normalizer.p -> Double.PositiveInfinity)
println("Normalized using L^inf norm")
lInfNormData.show()

// 向量的无穷阶范数即向量中所有值中的最大值
+---+-------------------+----------------------+
| id|  features   |  normFeatures|
+---+------------------+-----------------------+
|  0|[1.0,0.5,-1.0]|[1.0,0.5,-1.0]   |
|  1| [2.0,1.0,1.0]| [1.0,0.5,0.5]   |
|  2|[4.0,10.0,2.0]| [0.4,1.0,0.2]   |
+---+-------------------+----------------------+
           

• 标准差规范器：StandardNormalizer

  将特征标准化为单位标准差或是0均值，或是0均值单位标准差。

  将每一列的标准差限制在0-1之间，从而倒推各特征值的缩放

  标准差公式：

  

import org.apache.spark.ml.feature.StandardScaler

val scaler = new StandardScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")
  .setWithStd(true)
  .setWithMean(false)

// Compute summary statistics by fitting the StandardScaler.
val scalerModel = scaler.fit(dataFrame)

// Normalize each feature to have unit standard deviation.
val scaledData = scalerModel.transform(dataFrame)
scaledData.show

// 将每一列的标准差缩放到1。
+---+--------------+------------------------------------------------------------+
|id |features      |scaledFeatures                                              |
+---+--------------+------------------------------------------------------------+
|0  |[1.0,0.5,-1.0]|[0.6546536707079772,0.09352195295828244,-0.6546536707079771]|
|1  |[2.0,1.0,1.0] |[1.3093073414159544,0.1870439059165649,0.6546536707079771]  |
|2  |[4.0,10.0,2.0]|[2.618614682831909,1.870439059165649,1.3093073414159542]    |
+---+--------------+------------------------------------------------------------+
           

• 值域范围缩放器：MinMaxScaler

  将一列特征的（| 最大值-最小值 |）作为分母，（特征x-最小值）作为分子

import org.apache.spark.ml.feature.MinMaxScaler

val scaler = new MinMaxScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")

// Compute summary statistics and generate MinMaxScalerModel
val scalerModel = scaler.fit(dataFrame)

// rescale each feature to range [min, max].
val scaledData = scalerModel.transform(dataFrame)
println(s"Features scaled to range: [${scaler.getMin}, ${scaler.getMax}]")
scaledData.select("features", "scaledFeatures").show

// 每维特征线性地映射，最小值映射到0，最大值映射到1。
+--------------+-----------------------------------------------------------+
|features      |scaledFeatures                                             |
+--------------+-----------------------------------------------------------+
|[1.0,0.5,-1.0]|[0.0,0.0,0.0]                                              |
|[2.0,1.0,1.0] |[0.3333333333333333,0.05263157894736842,0.6666666666666666]|
|[4.0,10.0,2.0]|[1.0,1.0,1.0]                                              |
+--------------+-----------------------------------------------------------+
           

• 最大绝对值缩放器：MaxAbsScaler

  将一列特征的最大绝对值作为分母，（特征x）作为分子

import org.apache.spark.ml.feature.MaxAbsScaler

val scaler = new MaxAbsScaler()
  .setInputCol("features")
  .setOutputCol("scaledFeatures")

// Compute summary statistics and generate MaxAbsScalerModel
val scalerModel = scaler.fit(dataFrame)

// rescale each feature to range [-1, 1]
val scaledData = scalerModel.transform(dataFrame)
scaledData.select("features", "scaledFeatures").show()

// 每一维的绝对值的最大值为[4, 10, 2]
+--------------+----------------+                                               
|      features|  scaledFeatures|
+--------------+----------------+
|[1.0,0.5,-1.0]|[0.25,0.05,-0.5]|
| [2.0,1.0,1.0]|   [0.5,0.1,0.5]|
|[4.0,10.0,2.0]|   [1.0,1.0,1.0]|
+--------------+----------------+
           

上述介绍了MLlib的底层架构，向量和矩阵编程接口，还有特征向量规范化处理

2. 模型评估

在机器学习中，如何评估一个训练出来的模型是否好，有多好是需要尽可能数据来量化的，这时候就需要使用算法来评估以及量化模型的准确程度

2.1 混淆矩阵

1. 矩阵，可以理解为就是一张表格，混淆矩阵其实就是一张表格而已。

以分类模型中最简单的二分类为例，对于这种问题，我们的模型最终需要判断样本的结果是0还是1，或者说是positive还是negative。

我们通过样本的采集，能够直接知道真实情况下，哪些数据结果是positive，哪些结果是negative。同时，我们通过用样本数据跑出分类型模型的结果，也可以知道模型认为这些数据哪些是positive，哪些是negative。

我们就能得到这样四个基础指标，我称他们是一级指标（最底层的）：

真实值是positive，模型认为是positive的数量（True Positive=TP）

真实值是positive，模型认为是negative的数量（False Negative=FN）：这就是统计学上的第一类错误（Type I Error）

真实值是negative，模型认为是positive的数量（False Positive=FP）：这就是统计学上的第二类错误（Type II Error）

真实值是negative，模型认为是negative的数量（True Negative=TN）

1. 预测性分类模型，肯定是希望越准越好。那么，对应到混淆矩阵中，那肯定是希望TP与TN的数量大，而FP与FN的数量小。所以当我们得到了模型的混淆矩阵后，就需要去看有多少观测值在第二、四象限对应的位置，这里的数值越多越好；反之，在第一、三象限对应位置出现的观测值肯定是越少越好。
2. 混淆矩阵里面统计的是个数，有时候面对大量的数据，光凭算个数，很难衡量模型的优劣。因此混淆矩阵在基本的统计结果上又延伸了如下4个指标，我称他们是二级指标（通过最底层指标加减乘除得到的）：

   

   准确率（Accuracy）—— 针对整个模型

   精确率（Precision）

   灵敏度（Sensitivity）：就是召回率（Recall）

   特异度（Specificity）

// 模型评估 混淆矩阵
val rdd = prediction.rdd.map(row=>{
  val label = row.getAs[Double]("label")
  val prediction = row.getAs[Double]("prediction")
  (prediction,label)
})
val matrix: Matrix = new MulticlassMetrics(rdd).confusionMatrix
           

2.2 AUC曲线(分类算法)

1. ROC曲线，即为一条ROC曲线（该曲线的原始数据第三部分会介绍）。现在关心的是：

   横轴：False Positive Rate（假阳率，FPR）

   纵轴：True Positive Rate（真阳率，TPR）

   ROC 就是假阳率和真阳率之间的关系曲线

   

假阳率，简单通俗来理解就是预测为正样本但是预测错了的可能性，显然，我们不希望该指标太高。

真阳率，则是代表预测为正样本但是预测对了的可能性，当然，我们希望真阳率越高越好。

显然，ROC曲线的横纵坐标都在[0,1]之间，自然ROC曲线的面积不大于1。现在我们来分析几个特殊情况，从而更好地掌握ROC曲线的性质：

(0,0)：假阳率和真阳率都为0，即分类器全部预测成负样本

(0,1)：假阳率为0，真阳率为1，全部完美预测正确，HAPPY

(1,0)：假阳率为1，真阳率为0，全部完美预测错误，悲剧

(1,1)：假阳率和真阳率都为1，即分类器全部预测成正样本

TPR＝FPR，斜对角线，预测为正样本的结果一半是对的，一半是错的，代表随机分类器的预测效果

于是，我们可以得到基本的结论：ROC曲线在斜对角线以下，则表示该分类器效果差于随机分类器，反之，效果好于随机分类器，当然，我们希望ROC曲线尽量除于斜对角线以上，也就是向左上角（0,1）凸。

1. AUC实际上就是ROC曲线下的面积

   ROC曲线一定程度上可以反映分类器的分类效果，但是不够直观，我们希望有这么一个指标，如果这个指标越大越好，越小越差，于是，就有了AUC

AUC直观地反映了ROC曲线表达的分类能力。

AUC ＝ 1，代表完美分类器

0.5 < AUC < 1，优于随机分类器

0 < AUC < 0.5，差于随机分类器

// 模型评估
val ev = new BinaryClassificationEvaluator()
    .setLabelCol("label")
  .setMetricName("areaUnderROC")
val roc: Double = ev.evaluate(prediction)
println(roc)
           

2.3 回归算法评估

注意，回归算法其实就是找出已有数据中，数据输入和输出之间关系，简单如线性回归，就是输入和输出之间的线性关系。

回归算法的输出结果，不可能用准确率精确率等指标来衡量

应该用预测值和真实值之间的误差情况来衡量，具体来说，通常有如下几个指标：

1. 回归分析评估指标

• RMSE（Root Mean Square Error）均方根误差

  衡量观测值与真实值之间的偏差。

  常用来作为机器学习模型预测结果衡量的标准

  

  开根号： 为了让量纲更直观

  100元,90元 => RMSE =10元

• MSE（Mean Square Error）均方误差

  MSE是真实值与预测值的差值的平方然后求和平均。

  通过平方的形式便于求导，所以常被用作线性回归的损失函数

  

  100元,90元 => MSE =100
• R2

  

• MAE（Mean Absolute Error）平均绝对误差

  是绝对误差的平均值。

  可以更好地反映预测值误差的实际情况。

  

2.4回归分析评估指标含义

• SSE

  预测值和真值相差的平方和是SSE,也就是误差平方和，这肯定是越小越好，相当于一个误差累计。当然这个SSE越接近于0越好

  

  但如果10000的样本的情况，建立一个A模型，这个模型的SSE是100，100个样本的情况下，建立一个B模型，这个模型的SSE是80。但是不能说B模型比A模型好。所以就引入了MSE
• MSE

  

  就是均方误差，SSE除以样本量，平均的预测的值和真值差的平方，平均到每一个预测的Y

  

• RMSE 对MSE开方就是RMSE，也就是均方根
• SSR 表示的是预测值和原始值得均值差得平方和

  

• SST 表示得是原始数据和均值的差的平方和

  

• R2 就是R-square，可以经过公式推导得出SST=SSE+SSR

  

  将R2写开

  

R2的范围在0-1之间，越接近1，表示越好

一般衡量线性回归最好的指标应该就是R2，通常表示模型拟合的好坏。

对R2开根号，就是R，也就是相关系数，也是越近1越好

上面真值和预测值之间的误差都是做差求平方和，如果将平方和换成取绝对值，也就是MAE,RMAE,也就是不是square，变为absolute

val evaluator4 = new RegressionEvaluator()
  .setLabelCol("label")
  .setPredictionCol("prediction")
  .setMetricName("mae")
val d4: Double = evaluator4.evaluate(prediction)
           

3. 总结

1. 本文主要是spark mllib模块简介v，包括mllib的底层架构，向量和矩阵编程接口，如何使用mllib做特征向量规范化处理
2. 当模型选择并训练好之后，如何进行模型的效果评估，直接基于训练集或者验证集并无法得出一个量化结果，这时候就需要使用算法进行评估。这些评估算法区分算法类型，回归类型，分类类型等
3. 回归分析评估指标较多，含义做了单独解释。注意，如果没有机器学习相关知识储备，不理解或者无法深刻理解这些指标和算法是很正常，因为我也遇到了同样问题，需要补充理论知识同时结合代码做实践，这样理解会更加深刻。