江苏省交通运输厅门户网站建设管理办法,建材 东莞网站建设,重庆网站供奉战犯,广东深圳住房和城乡建设部网站SparkML
机器学习: 让机器学会人的学习行为, 通过算法和数据来模拟或实现人类的学习行为#xff0c;使之不断改善自身性能。
机器学习的步骤: 加载数据特征工程 数据筛选: 选取适合训练的特征列, 例如用户id就不适合, 因为它特性太显著.数据转化: 将字符串的数据转化数据类型…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 PipelinesPersistence持久化模型的保存、读取、管道操作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(inputColcategory, outputColcategoryIndex)# 先调用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(inputColcategory, outputColcategoryIndex)# 先调用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(inputColcategoryIndex, outputColcategoryString)# 直接调用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(inputColfeatures, outputColscaledFeatures)# 调用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], outputColfeatures)vectorAssembler.transform(df).show(truncateFalse)# -------------------------------------------------------------# |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值], outputColfeatures)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-meanskmeans KMeans(featuresColfeatures, k2, predictionColprediction, seed125)# 先调用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(predictionColprediction, featuresColfeatures)# 计算轮廓系数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 拆分为子集使用该特征信息增益最大的特征作为决策树的一个节点使用剩余特征对子集重复上述123过程 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], seed123)# 模型训练
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(predictionColprediction, labelCollabel)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 , 加载模型后 ,对后续数据进行分类会很简单
