SparkML机器学习

SparkML

机器学习: 让机器学会人的学习行为, 通过算法和数据来模拟或实现人类的学习行为，使之不断改善自身性能。

机器学习的步骤:

• 加载数据
• 特征工程
  • 数据筛选: 选取适合训练的特征列, 例如用户id就不适合, 因为它特性太显著.
  • 数据转化:
    • 将字符串的数据转化数据类型, 因为模型训练的数据不能为字符串.
    • 将多个特征列转化为一个向量列, 因为spark机器学习要求数据输入只能为一个特征列
  • 数据缩放:
    • 把所有的特征缩放到0~1之间，都处于相同的量纲大小范围内.
    • 公式:（特征值- 当前特征最小值）/ （当前特征最大值- 当前特征最小值）
• 模型训练
  • 创建模型: 机器学习算法对象.fit(train_df) 以此来创建模型
  • 预测数据: 模型.transform(test_df) 多返回一个预测列
• 模型评估
  • 有监督 :有监督学习是指训练模型时使用的训练数据是标识好的数据, 也就是说数据集既有特征列也有目标列.
    • 分类问题(目标值为离散的): 预测数据和目标值进行对比, 查看预测成功的几率
    • 回归问题(目标值为连续的): 求测试集预测结果与目标值标准差：方差开根号
  • 无监督KMeans的 :
    • sse:表示数据样本与它所属的簇中心之间的距离（差异度）平方之和
    • sc: (b-a)/max(a,b)
      • a: 样本i到同一簇内其他点不相似程度(欧式距离,余弦定理…)的平均值
      • b: 样本I到其他簇的平均不相似度的最小值
• 模型上线: 模型的保存和加载

MLLib机器学习库简介:

MLLIB是Spark的机器学习库。提供了利用Spark构建大规模和易用性的机器学习平台，组件：

• ML 算法：包括了分类、回归、聚类、降维、协同过滤
• Featurization特征化：特征抽取、特征转换、特征降维、特征选择
• Pipelines管道：tools for constructing, evaluating, and tuning ML Pipelines
• Persistence持久化：模型的保存、读取、管道操作
• Utilities：提供了线性代数、统计学以及数据处理工具

数据加载

因为现在所用的ml库只支持Dataframe格式的数据, 因此需要加载df的数据, 例如读取csv文件:

data_df = spark.read.format('csv') \.option('header', True) \.option('inferSchema', True) \.load('./a.txt')

特征工程

StringIndexer

将字符串类型转化为数值类型, 因为机器学习要求输入必须是数值类型.

如果有多列需要转化, 则需要写多个StringIndexer函数, 下一个函数的输入是这个函数的输出.

from pyspark.ml.feature import StringIndexer
from pyspark.sql import SparkSessionif __name__ == '__main__':# 创建spark运行环境spark = SparkSession.builder.appName("stringIndexer test") \.master('local[*]').getOrCreate()# 创建测试df#  id | category# ----|----------#  0  | a#  1  | b#  2  | c#  3  | a#  4  | a#  5  | cdf = spark.createDataFrame([(0, "a"), (1, "b"), (2, "c"), (3, "a"), (4, "a"), (5, "c")], ["id", "category"])# 将df中的字符串列转化数值列stringIndexer = StringIndexer(inputCol='category', outputCol='categoryIndex')# 先调用fit方法创建模型model = stringIndexer.fit(df)#　再调用transform方法进行模型计算返回结果result_df = model.transform(df)result_df.show()# +---+--------+-------------+# | id|category|categoryIndex|# +---+--------+-------------+# |  0|       a|          0.0|# |  1|       b|          2.0|# |  2|       c|          1.0|# |  3|       a|          0.0|# |  4|       a|          0.0|# |  5|       c|          1.0|# +---+--------+-------------+# 根据字符出再频数排序（倒序），分开转化为[0.0, 1.0, 2.0, ....., numLabels]

IndexToString

与StringIndexer结合使用，IndexToString将一列标签索引映射回包含原始标签作为字符串的列。

一个常见的用例是从标签生成索引StringIndexer，使用这些索引训练模型，并从预测索引列中检索原始标签IndexToString。

from pyspark.ml.feature import StringIndexer, IndexToString
from pyspark.sql import SparkSessionif __name__ == '__main__':# 创建spark运行环境spark = SparkSession.builder.appName("stringIndexer test") \.master('local[*]').getOrCreate()# 创建测试df#  id | category# ----|----------#  0  | a#  1  | b#  2  | c#  3  | a#  4  | a#  5  | cdf = spark.createDataFrame([(0, "a"), (1, "b"), (2, "c"), (3, "a"), (4, "a"), (5, "c")], ["id", "category"])# 将df中的字符串列转化数值列stringIndexer = StringIndexer(inputCol='category', outputCol='categoryIndex')# 先调用fit方法创建模型model = stringIndexer.fit(df)#　再调用transform方法进行模型计算返回结果result_df = model.transform(df)result_df.show()# +---+--------+-------------+# | id|category|categoryIndex|# +---+--------+-------------+# |  0|       a|          0.0|# |  1|       b|          2.0|# |  2|       c|          1.0|# |  3|       a|          0.0|# |  4|       a|          0.0|# |  5|       c|          1.0|# +---+--------+-------------+# 根据字符出再频数排序（倒序），分开转化为[0.0, 1.0, 2.0, ....., numLabels]index2String = IndexToString(inputCol='categoryIndex', outputCol='categoryString')# 直接调用transform计算结果string_result_df = index2String.transform(result_df)string_result_df.show()# +---+--------+-------------+--------------+# | id|category|categoryIndex|categoryString|# +---+--------+-------------+--------------+# |  0|       a|          0.0|             a|# |  1|       b|          2.0|             b|# |  2|       c|          1.0|             c|# |  3|       a|          0.0|             a|# |  4|       a|          0.0|             a|# |  5|       c|          1.0|             c|# +---+--------+-------------+--------------+

MinmaxScaler

特征缩放（归一化处理）：（特征值- 当前特征最小值）/ （当前特征最大值- 当前特征最小值）

• 作用：把所有的特征缩放到0~1之间，都处于相同的量纲大小范围内
• 场景：当特征之间量纲（取值大小）差距比较大的时候，需要先做特征缩放（归一化）

from pyspark.ml.feature import MinMaxScaler
from pyspark.ml.linalg import Vectors
from pyspark.sql import SparkSessionif __name__ == '__main__':# 创建spark运行环境spark = SparkSession.builder.appName("stringIndexer test") \.master('local[*]').getOrCreate()# 创建测试DFdataFrame = spark.createDataFrame([(0, Vectors.dense([1.0, 0.1, -1.0]),),(1, Vectors.dense([2.0, 1.1, 1.0]),),(2, Vectors.dense([3.0, 10.1, 3.0]),)], ["id", "features"])# 对特征列进行特征缩放minmaxScaler = MinMaxScaler(inputCol='features', outputCol='scaledFeatures')# 调用fit方法创建模型，调用transform方法进行模型计算result_df = minmaxScaler.fit(dataFrame).transform(dataFrame)result_df.show()# +---+--------------+--------------+# | id|      features|scaledFeatures|# +---+--------------+--------------+# |  0|[1.0,0.1,-1.0]| [0.0,0.0,0.0]|# |  1| [2.0,1.1,1.0]| [0.5,0.1,0.5]|# |  2|[3.0,10.1,3.0]| [1.0,1.0,1.0]|# +---+--------------+--------------+

VectorAssembler

• 作用：VectorAssembler是一个变换器，它将给定的列表组合到一个向量列中。
• 场景：在模型训练前, 我们要把模型训练用到的所有特征， 都放到一列中, 并且需要是向量的形式,，这个是Spark MLlib的硬性要求(模型训练传数据的时候, 只接收一列数据)，只要涉及模型训练就会使用到VectorAssember。

from pyspark.ml.clustering import KMeans
from pyspark.ml.feature import MinMaxScaler, VectorAssembler
from pyspark.ml.linalg import Vectors
from pyspark.sql import SparkSessionif __name__ == '__main__':# 创建spark运行环境spark = SparkSession.builder.appName("stringIndexer test") \.master('local[*]').getOrCreate()# 创建测试数据df = spark.createDataFrame([(0, 18, 1.0, Vectors.dense([0.0, 10.0, 0.5]), 1.0)],["id", "hour", "mobile", "userFeatures", "clicked"])# 把模型计算需要的特征列都封装到一个向量列中vectorAssembler = VectorAssembler(inputCols=['hour', 'mobile', 'userFeatures', 'clicked'], outputCol='features')vectorAssembler.transform(df).show(truncate=False)# +---+----+------+--------------+-------+---------------------------+# |id |hour|mobile|userFeatures  |clicked|features                   |# +---+----+------+--------------+-------+---------------------------+# |0  |18  |1.0   |[0.0,10.0,0.5]|1.0    |[18.0,1.0,0.0,10.0,0.5,1.0]|# +---+----+------+--------------+-------+---------------------------+

模型训练和评估

KMeans无监督聚类算法

算法原理:

k-means其实包含两层内容：K表示初始中心点个数（计划聚类数），means求中心点到其他数据点距离的平均值, K自己设置(2,3,4,5,6,7,8)

具体步骤如下：

• 随机选取K个中心点
• 求每个元素点和每个中心点的欧氏距离
• 如果元素点离某个中心点距离最近, 则归属与该中心点(簇cluster)
• 求每个簇中所有元素点的平均值, 以此作为新的一个中心点
• 重复2-4步骤, 进行迭代计算, 直到达到终止迭代条件
  • 最大迭代次数,默认是20次
  • 本次中心点和上次中心点重合(这里的重合不是指完全重合, 存在一定的误差范围)

特点:

优点：速度快，简单

• 对处理大数据集，该算法保持可伸缩性和高效率。
• 当簇近似为高斯分布时，它的效果较好。

缺点：最终结果跟初始点选择相关，容易陷入局部最优

• k均值算法中k是实现者给定的，这个k值的选定是非常难估计的。
• k均值的聚类算法需要不断地进行样本分类调整，不断地计算调整后的新的聚类中心，当数据量大的时候，算法开销很大。
• k均值是求得局部最优解的算法，所以对于初始化时选取的k个聚类的中心比较敏感，不同点的中心选取策略可能带来不同的聚类结果。
• 对噪声点和孤立点数据敏感。

KMeans一般是其他聚类方法的基础算法，如谱聚类。

评估方法:

• sse:表示数据样本与它所属的簇中心之间的距离（差异度）平方之和
• sc: (b-a)/max(a,b)
  • a: 样本i到同一簇内其他点不相似程度(欧式距离,余弦定理…)的平均值
  • b: 样本I到其他簇的平均不相似度的最小值

from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
from pyspark.ml.feature import VectorAssembler
from pyspark.sql import SparkSessionif __name__ == '__main__':# 创建spark运行环境spark = SparkSession.builder \.appName("kmeans 算法实现") \.master('local[*]') \.getOrCreate()# ====================== 1、加载数据 ===================data_df = spark.read.format('csv') \.option('header', True) \.option('inferSchema', True) \.load('/tmp/pycharm_project_488/com/itheima/data/test.txt')# data_df.show()# +-----------+----+# |Weightindex|PH值|# +-----------+----+# |        1.0| 1.0|# |        2.0| 1.0|# |        4.0| 3.0|# |        5.0| 4.0|# +-----------+----+# ===================== 2、特征工程 =======================# 将特征列封装到一个向量列中assembler = VectorAssembler(inputCols=['Weightindex', 'PH值'], outputCol='features')vector_df = assembler.transform(data_df)# vector_df.show()# +-----------+----+---------+# |Weightindex|PH值| features|# +-----------+----+---------+# |        1.0| 1.0|[1.0,1.0]|# |        2.0| 1.0|[2.0,1.0]|# |        4.0| 3.0|[4.0,3.0]|# |        5.0| 4.0|[5.0,4.0]|# +-----------+----+---------+# ===================== 3、模型训练 =======================# featuresCol 设置特征列# k 设置簇数量# predictionCol 设置目标列# seed 随机种子# initMode: str = 'k-means||' k-means++kmeans = KMeans(featuresCol='features', k=2, predictionCol='prediction', seed=125)# 先调用fit方法创建模型model = kmeans.fit(vector_df)# 再调用transform方法计算结果result_df = model.transform(vector_df)# result_df.show()# +-----------+----+---------+----------+# |Weightindex|PH值| features|prediction|# +-----------+----+---------+----------+# |        1.0| 1.0|[1.0,1.0]|         0|# |        2.0| 1.0|[2.0,1.0]|         0|# |        4.0| 3.0|[4.0,3.0]|         1|# |        5.0| 4.0|[5.0,4.0]|         1|# +-----------+----+---------+----------+# ======================== 4、模型评估 ====================evaluator = ClusteringEvaluator(predictionCol='prediction', featuresCol='features')# 计算轮廓系数sc = evaluator.evaluate(result_df)print("轮廓系数：", sc) # 轮廓系数： 0.8967364744598525# 获取中心点信息centers = model.clusterCenters()print("中心点信息：", centers) # 中心点信息： [array([1.5, 1. ]), array([4.5, 3.5])]# 中心点可以是计算得到的虚拟点

决策树有监督分类问题算法

概述:

决策树算法是一种监督学习算法，英文是Decision tree。

决策树是一个类似于流程图的树结构：其中，每个内部结点表示一个特征或属性，而每个树叶结点代表一个分类。树的最顶层是根结点。使用决策树分类时就是将实例分配到叶节点的类中。该叶节点所属的类就是该节点的分类。

决策树思想的来源非常朴素，试想每个人的大脑都有类似于if-else这样的逻辑判断，这其中的if表示的是条件，if之后的then就是一种选择或决策。程序设计中的条件分支结构就是if-then结构，最早的决策树就是利用这类结构分割数据的一种分类学习方法。

生成步骤:

构建决策树包括三个步骤：

特征选择：选取有较强分类能力的特征。

决策树生成：典型的算法有ID3、C4.5、CART，它们生成决策树过程相似，ID3是采用信息增益作为特征选择度量，而C4.5采用信息增益率、CART基尼指数。

决策树剪枝：剪枝原因是决策树生成算法生成的树对训练数据的预测很准确，但是对于未知数据分类很差，这就产生了过拟合的现象。

• 过拟合：训练集表现较好，但是测试集表现不好可以通过剪枝（减少特征列）
• 欠拟合：训练集表现不好，测试集表现不好，通过增加特征列解决

算法介绍:

D3算法步骤：

• 计算每个特征的信息增益=经验熵-条件熵：整个数据集的信息熵-当前节点的信息熵
• 使用信息增益最大的特征将数据集 S 拆分为子集
• 使用该特征（信息增益最大的特征）作为决策树的一个节点
• 使用剩余特征对子集重复上述（1，2，3）过程

C4.5算法介绍：

• C4.5 是计算信息增益率 ：信息增益/当前特征取值的信息熵

• 解决ID3决策树缺点

  • 当前特征列的取值越多时，信息增益越大
  • ID3会偏向于选择特征列取值比较多的特征列

cart模型算法:

Cart模型是一种决策树模型，它即可以用于分类，也可以用于回归，其学习算法分为下面两步：

（1）决策树生成：用训练数据生成决策树，生成树尽可能大。

（2）决策树剪枝：基于损失函数最小化的剪枝，用验证数据对生成的数据进行剪枝。

Cart算法通过计算基尼指数(GINI)来选择特征：基尼指数=1-∑Pi²

• 信息增益（ID3）、信息增益率（C4.5）值越大，则说明优先选择该特征。
• 基尼指数值越小（cart），则说明优先选择该特征。

模型评估:

将测试的预测列正确的数/测试的数据条数 = 预测正确率

from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.feature import StringIndexer, VectorAssembler
from pyspark.sql import SparkSessionif __name__ == '__main__':
# 创建spark运行环境
spark = SparkSession.builder.getOrCreate()# 构建测试数据df
iris_df = spark.read.format('csv')\.option('header', True)\.option('inferSchema', True)\.option('sep', ',')\.load("/tmp/pycharm_project_488/com/itheima/data/iris.csv")iris_df.show()# 特征工程-特征转换
indexer = StringIndexer().setInputCol("class").setOutputCol("label")indexer_df = indexer.fit(iris_df).transform(iris_df)indexer_df.show()
# 特征工程-特征选择
assembler = VectorAssembler().setInputCols(['sepal_length', 'sepal_width', 'petal_length', 'petal_width']).setOutputCol('features')vector_df = assembler.transform(indexer_df)vector_df.show()# 切分数据集
(train_df, test_df) = vector_df.randomSplit([0.8, 0.2], seed=123)# 模型训练
classifier = DecisionTreeClassifier() \.setFeaturesCol('features') \.setLabelCol('label') \.setPredictionCol('prediction') \.setMaxDepth(4) \.setImpurity('gini')# 通过训练数据集构建模型
model = classifier.fit(train_df)# 使用模型对训练数据集进行预测计算
train_result = model.transform(train_df)
train_result.show()
# 使用模型对测试数据集进行预测计算
test_result = model.transform(test_df)
test_result.show()# 模型评估
print("训练集的准确率：", (train_result.filter("label == prediction").count() / train_result.count()))
print("测试集的准确率：", (test_result.filter("label == prediction").count() / test_result.count()))evaluator = MulticlassClassificationEvaluator(predictionCol='prediction', labelCol='label')print("训练集的准确率", evaluator.evaluate(train_result))
print("测试集的准确率", evaluator.evaluate(test_result))

模型上线

Pipeline介绍:从Spark1.2版本之后引入了ML Pipeline，经过多个版本的发展，SparkMl克服了Mllib在处理复杂机器学习问题的一些不足，如工作比较复杂，流程不够清晰等，向用户提供基于DataFrame之上的更高层次的API库，以方便的构建复杂的机器学习工作流式应用，使得整个机器学习构建过程更加简单、高效和规范。

Pipeline功能:

• 减少代码量
• 流程更清晰
• pipeline中所有stage的入参都是相同

pipeline = Pipeline().setStages([indexer, assembler, classifier])
model = pipeline.fit(iris_df)
result_df = model.transform(iris_df)
result_df.show()

模型保存:

model.save('hdfs:///model/usegmodel')
# 模型保存时路径下要求没有文件, 如果有文件会报错

模型加载:

model = PipelineModel.load('hdfs:///model/usegmodel')
result_df2 = model.transform(source_df)
result_df2.show()
# 如果能使用Pipeline 尽量使用Pipeline, 保存一个Pipelinemodel , 加载模型后 ,对后续数据进行分类会很简单