14.spark mllib之快速入门

Posted 2021-01-05

简介

??MLlib是Spark提供提供机器学习的库，专为在集群上并行运行的情况而设计。
MLlib包含很多机器学习算法，可在Spark支持的所有编程语言中使用。

??MLlib设计理念是将数据以RDD的形式表示，然后在分布式数据集上调用各种算法。其实，MLlib就是RDD上一系列可供调用的函数的集合。

数据类型

??MLlib包含一些特有的数据类型，位于org.apache.spark.mllib包（Java/Scala）或pyspark.mllib（Python）中。主要的几个类有：

• Vector

  • 一个本地向量（Local Vector）。索引是从0开始的，并且是整型。而值为 Double 类型，存储于单个机器内。

  • MLlib既支持稠密向量也支持稀疏向量，前者表示向量的每一位都存储，后者只存储非零位以节约空间。

  • 向量可以通过mllib.linalg.Vectors类创建

  • scala

  //创建稠密向量
  scala> val denseVec1 = Vectors.dense(1.0,2.0,3.0)
  denseVec1: org.apache.spark.mllib.linalg.Vector = [1.0,2.0,3.0]
  
  scala> val denseVec2 = Vectors.dense(Array(1.0,2.0,3.0))
  denseVec2: org.apache.spark.mllib.linalg.Vector = [1.0,2.0,3.0]
  
  //创建稀疏向量
  scala> val sparseVec1 = Vectors.sparse(4,Array(0,2),Array(1.0,2.0))
  sparseVec1: org.apache.spark.mllib.linalg.Vector = (4,[0,2],[1.0,2.0])

  • python

  >>> from pyspark.mllib.linalg import Vectors
  >>> den = Vectors.dense([1.0,2.0,3.0])
  >>> den
  DenseVector([1.0, 2.0, 3.0])
  >>> spa = Vectors.sparse(4,[0,2],[1.0,2.0])
  >>> spa
  SparseVector(4, {0: 1.0, 2: 2.0})

• LabeledPoint

  • 在分类和回归之类的监督式学习（supervised learning）算法中使用。

  • LabeledPoint表示带标签的数据点，包括一个特征向量与一个标签（由一个浮点数表示）。

  • 位于mllib.regression包中

  • scala

  // 首先需要引入标签点相关的类
  import org.apache.spark.mllib.linalg.Vectors
  import org.apache.spark.mllib.regression.LabeledPoint
  
  // 创建一个带有正面标签和稠密特征向量的标签点。
  val pos = LabeledPoint(1.0, Vectors.dense(1.0, 0.0, 3.0))
  
  // 创建一个带有负面标签和稀疏特征向量的标签点。
  val neg = LabeledPoint(0.0, Vectors.sparse(3, Array(0, 2), Array(1.0, 3.0)))

  • python

  >>> from pyspark.mllib.regression import LabeledPoint
  >>> from pyspark.mllib.linalg import Vectors
  >>> pos = LabeledPoint(1.0,Vectors.dense([1.0,2.0,3.0]))
  >>> neg = LabeledPoint(0.0,Vectors.dense([1.0,2.0,3.0]))

• Matrix

  • 矩阵分为稠密矩阵和稀疏矩阵

  • 稠密矩阵的实体值以列为主要次序的形式，存放于单个 Double 型数组内。系数矩阵的非零实体以列为主要次序的形式，存放于压缩稀疏列（Compressed Sparse Column, CSC）中。例如，下面这个稠密矩阵就是存放在一维数组 [1.0, 3.0, 5.0, 2.0, 4.0, 6.0] 中，矩阵的大小为 (3, 2) 。

  • 本地矩阵的基类是 Matrix 类，在 Spark 中有其两种实现，分别是 DenseMatrix 和 SparseMatrix 。官方文档中推荐使用 已在 Matrices 类中实现的工厂方法来创建本地矩阵。需要注意的是，MLlib 中的本地矩阵是列主序的（column-major）

  • 稠密矩阵

    import org.apache.spark.mllib.linalg.{Matrix, Matrices}
    // 创建稠密矩阵 ((1.0, 2.0), (3.0, 4.0), (5.0, 6.0))
    val dm: Matrix = Matrices.dense(3, 2, Array(1.0, 3.0, 5.0, 2.0, 4.0, 6.0))

  • 稀疏矩阵

    scala> val sparseMatrix= Matrices.sparse(3, 3, Array(0, 2, 3, 6), Array(0, 2, 1, 0, 1, 2), Array(1.0, 2.0, 3.0,4.0,5.0,6.0))
    sparseMatrix: org.apache.spark.mllib.linalg.Matrix = 
    3 x 3 CSCMatrix
    (0,0) 1.0
    (2,0) 2.0
    (1,1) 3.0
    (0,2) 4.0
    (1,2) 5.0
    (2,2) 6.0

• Rating

  • 用于产品推荐

  • 表示用户对一个产品的评分

  • 位于mllib.recommendation包中

• 各种Model类（模型）

  • 每个Model都是训练算法的结果

  • 模型一般都有一个predict()方法，使用该模型对新的数据点或数据点组成的RDD进行预测。

统计

??不论是在即时的探索中，还是在机器学习的数据理解中，基本的统计都是数据分析的重要部分。MLlib 通过mllib.stat.Statistics 类中的方法提供了几种广泛使用的统计函数，这些函数可以直接在RDD 上使用。一些常用的函数如下所列。

Statistics.colStats(rdd)

??计算由向量组成的RDD 的汇总统计，保存着向量集合中每列的最小值、最大值、平均值和方差。这可以用来在一次执行中获取丰富的统计信息。

Statistics.corr(rdd, method)

?&esmp;计算由向量组成的RDD 中的列间的相关矩阵，使用皮尔森相关（Pearson correlation）或斯皮尔曼相关（Spearman correlation）中的一种（method 必须是pearson 或spearman中的一个）。

Statistics.corr(rdd1, rdd2, method)

??计算两个由浮点值组成的RDD 的相关矩阵，使用皮尔森相关或斯皮尔曼相关中的一种（method 必须是pearson 或spearman 中的一个）。

Statistics.chiSqTest(rdd)

??计算由LabeledPoint 对象组成的RDD 中每个特征与标签的皮尔森独立性测试
（Pearson’s independence test） 结果。返回一个ChiSqTestResult 对象， 其中有p 值、（p-value）、测试统计及每个特征的自由度。标签和特征值必须是分类的（即离散值）。

??下面举个例子：使用三个学生的成绩Vector来构建所需的RDD Vector，这个矩阵里的每个Vector都代表一个学生在四门课程里的分数：

python

from pyspark.mllib.stat import Statistics
from pyspark.mllib.linalg import Vectors
//构建RDD
basicTestRDD = sc.parallelize([Vectors.dense([60, 70, 80, 0]),
                       Vectors.dense([80, 50, 0,  90]),
                       Vectors.dense([60, 70, 80,  0])])

//查看summary里的成员，这个对象中包含了大量的统计内容
>>> print summary.mean()
[ 66.66666667  63.33333333  53.33333333  30.        ]
>>> print summary.variance()
[  133.33333333   133.33333333  2133.33333333  2700.        ]
>>> print summary.numNonzeros()
[ 3.  3.  2.  1.]

scala

import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.apache.spark.rdd.RDD

val array1: Array[Double] = Array[Double](60, 70, 80, 0)
val array2: Array[Double] = Array[Double](80, 50, 0, 90)
val array3: Array[Double] = Array[Double](60, 70, 80, 0)
val denseArray1 = Vectors.dense(array1)
val denseArray2 = Vectors.dense(array2)
val denseArray3 = Vectors.dense(array3)

val seqDenseArray: Seq[Vector] = Seq(denseArray1, denseArray2, denseArray3)

val basicTestRDD: RDD[Vector] = sc.parallelize[Vector](seqDenseArray)

val summary: MultivariateStatisticalSummary = Statistics.colStats(basicTestRDD)

算法

特征提取

• TF-IDF(词频——逆文档频率)是用来从文本文档（例如网页）中生成特定向量的简单方法。
• 缩放，大多数要考虑特征向量中各元素的幅值，并且在特征缩放调整为平等对待时表现最好。
• 规化，在准备输入数据时，把向量正规化为长度1。使用Normalizer类可以实现。
• Word2Vec是一个基于神经网络的文本特征算法，可以用来将数据传给许多下游算法。

降维

• 主成分分析（PCA）

  • PCA会把特征映射到低位空间，让数据在低维空间表示的方差最大化，从而忽略一些无用的维度。
  • 要计算这种映射，我们要构建出正规化的相关矩阵，并使用这个矩阵的奇异向量和奇异值。
  • 最大的一部分奇异值相对应的奇异向量可以用来重建原始数据的主要成分。
• 奇异值分解
  • MLlib也提供底层的奇异值分解（简称SVD）原语。

分类与回归

• 分类与回归是监督学习的两种形式。
• 监督学习是指算法尝试使用有标签的训练数据根据对象的特征预测结果。
• 在分类中，预测出的变量是离散的。
• 在回归中，预测出的变量是连续的。
• MLlib中包含许多分类与回归算法：如简单的线性算法以及决策树和森林算法。

聚类

• 聚类算法是一种无监督学习任务，用于将对象分到具有高度相似性的聚类中。
• 聚类算法主要用于数据探索（查看一个新数据集是什么样子）以及异常检测（识别与任意聚类都相聚较远的点）。
• MLlib中包含两个聚类中流行的K-means算法，以及一个叫做K-means||的变种，可以提供为并行环境提供更好的初始化策略。

协同过滤与推荐

• 协同过滤是一种根据用户对各种产品的交互与评分来推荐新产品的推荐系统技术。
• 交替最小二乘（ALS），会为每个用户和产品都设一个特征向量，这样用户向量和产品向量的点积就接近于他们的得分。

实例

使用逻辑回归算法实现垃圾邮件分类处理

def testLogisticRegressionWithSGD = {
    val spam = sc.textFile("src/main/resources/mllib/spam.txt", 1)
    val normal = sc.textFile("src/main/resources/mllib/normal.txt", 1)

    //创建一个HashingTF实例来把邮件文本映射为包含一个10000个特征的向量
    val tf = new HashingTF(numFeatures = 10000)
    //各邮件都被切分为单词，每个单词被映射为一个特征
    val spamFeatures = spam.map { email => tf.transform(email.split(" ")) }
    val normalFeatures = normal.map { email => tf.transform(email.split(" ")) }

    //创建LabeledPoint数据集分别存放阳性(垃圾邮件)和阴性(正常邮件)的例子
    val positiveExamples = spamFeatures.map { features => LabeledPoint(1, features) }
    val negativeExamples = normalFeatures.map { features => LabeledPoint(0, features) }
    val trainingData = positiveExamples.union(negativeExamples)
    trainingData.cache()
    println(trainingData.toDebugString)

    //使用SGD算法运行逻辑回归
    val model = new LogisticRegressionWithSGD().run(trainingData)
    //以阳性(垃圾邮件)和阴性(正常邮件)的例子分别进行测试
    val posTest = tf.transform("O M G get cheap stuff by sending money to .".split(" "))
    val negTest = tf.transform("hello, i started studying Spark ".split(" "))
    println(s"prediction for positive tset example: ${model.predict(posTest)}")
    println(s"prediction for negitive tset example: ${model.predict(negTest)}")

    Thread.sleep(Int.MaxValue)
  }

svm分类算法

# 加载模块
from pyspark.mllib.util import MLUtils
from pyspark.mllib.classification import SVMWithSGD

# 读取数据
dataFile = ‘/opt/spark-1.6.1-bin-hadoop2.6/data/mllib/sample_libsvm_data.txt‘
data = MLUtils.loadLibSVMFile(sc, dataFile)

splits = data.randomSplit([0.8, 0.2], seed = 9L)
training = splits[0].cache()
test = splits[1]

# 打印分割后的数据量
print "TrainingCount:[%d]" % training.count();
print "TestingCount:[%d]" % test.count();

model = SVMWithSGD.train(training, 100)

scoreAndLabels = test.map(lambda point : (model.predict(point.features), point.label))

#输出结果，包含预测的数字结果和0/1结果：
for score, label in scoreAndLabels.collect():
    print score, label

k-means聚类算法

# 读取数据文件，创建RDD
dataFile = "/opt/spark-1.6.1-bin-hadoop2.6/data/mllib/kmeans_data.txt"
lines = sc.textFile(dataFile)

# 创建Vector，将每行的数据用空格分隔后转成浮点值返回numpy的array
data = lines.map(lambda line: np.array([float(x) for x in line.split(‘ ‘)]))

# 其中2是簇的个数
model = KMeans.train(data, 2)

print("Final centers: " + str(model.clusterCenters))
print("Total Cost: " + str(model.computeCost(data)))