基于Spark的机器学习资料43、其它SparkML算法简单介绍

Spark ML（机器学习）介绍（监督学习、半监督学习、无监督学习）一、机器学习定义机器学习(Machine Learning, ML)是一门多领域交叉学科，涉及概率论、统计学、逼近论、凸分析、算法复杂度理论等多门学科。

专门研究计算机怎样模拟或实现人类的学习行为，以获取新的知识或技能，重新组织已有的知识结构使之不断改善自身的性能。

机器学习还有模式识别、计算机视觉、语音识别、统计学习以及自然语言处理等，而在目前机器学习中，深度学习是机器学习研究中的一个新的领域，其动机在于建立、模拟人脑进行分析学习的神经网络，它模仿人脑的机制来解释数据，例如图像，声音和文本。

机器学习代表人物：1、Andrew Ng （吴恩达）不久前刚从百度的首席数字科学家的职位离开，他还是斯坦福大学计算机科学系副教授，在线教育平台Coursera 的联合创始人之一。

2、Yoshua Bengio 是蒙特利尔大学计算机科学教授，他因在人工神经网络和深度学习方面的研究而著名。

Bengio 曾说，他的研究背后最大的野心是理解“获取智慧的学习的原则”。

除了AI 和ML 的其他原则之外，他发表的大部分工作设计用于编码或格式化非结构化数据的自动编码器，通过无监督机器学习令计算机可以理解。

3、Yann LeCun作为自2013年以来Facebook 的AI 研究主管，LeCun 在计算机视觉领域的开创性工作获得广泛认可，计算机视觉是教机器以与人类识别物体相同的方式去“看见”物体，并通过对它们进行分类的方式学习它们。

LeCun 也被认为是卷积神经网络的发明人之一，这类模型的目标是创建与生物有机体（例如眼睛或大脑）相同的方式去获取及解释信息的算法。

LeCun 是纽约大学数据科学中心的创始人。

4、Demis Hassabis 是DeepMind 联合创始人，DeepMind 是谷歌在2014年收购的一家英国AI 创业公司。

Hassabis 的工作重点是结合机器学习和神经科学方法促进人工神经网络的发展。

利用SparkML实现大数据分析与机器学习在当今信息时代，数据已经成为了不可或缺的资源。

特别是在企业建设和发展中，大数据已经成为了一个重要的资产。

然而，对于企业来说，如何处理这样大量的数据，以及如何从中获取更多的价值，却成为了一个难题。

这时，机器学习就出现了，它可以帮助企业快速发现有价值的信息。

而SparkML就是大数据分析与机器学习的一个非常有前途的工具。

一、Spark与SparkMLSpark是一种快速的分析大数据的工具。

它的核心特点是分布式数据处理，并且其基于内存而不是基于硬盘的运算方式，可以大幅度提高数据处理的速度。

同时，Spark还支持数据的流计算和批处理，并且可以对数据进行高斯模糊和机器学习。

Spark已经成为大数据领域中非常重要的一项技术。

SparkML是一种利用Spark进行大数据分析与机器学习的工具，是Spark生态系统中的一个子项目。

SparkML的特点是可以在分布式环境下同时处理多个大数据集，而且这些数据集可以运用不同的机器学习算法。

除此之外，SparkML还提供了数据转换、数据分析、数据挖掘和数据可视化等相关工具。

二、SparkML的机器学习工具SparkML提供了很多机器学习工具，主要包括以下几种：1.分类器：SparkML的分类器是基于Naive Bayes、支持向量机和决策树等算法的。

通过分析现有的数据集，分类器可以判断新的数据属于哪个类别，并给出相应的预测结果。

2.聚类器：SparkML的聚类器是基于K-means和Bisecting K-means等算法的。

聚类器可以将数据集中的数据分成不同的群组，每一组数据都有自己的一些共同特征。

3.回归器：SparkML的回归器是基于线性回归和逻辑回归等算法的。

回归器可以从已知的数据中预测新的未知数据的数值或者群组信息。

4.协同过滤器：SparkML的协同过滤器是基于推荐算法的。

它可以对大量的数据进行分析和处理，并且根据用户的偏好和历史行为，给出相应的推荐结果。

基于Spark的机器学习应用与算法实现探究在当前大数据时代，机器学习已经广泛应用于各个领域。

Spark作为一种快速、通用的大数据处理引擎，具有高效、灵活且易于使用的特点，成为了许多机器学习应用的首选平台。

本文将探究基于Spark的机器学习应用与算法实现的相关内容。

首先，我们来了解一下什么是Spark。

Spark是一种基于内存的分布式计算框架，提供了高效的数据处理能力，可以处理大规模数据，并且可以在多个节点上进行并行计算。

与其他分布式框架相比，Spark具有更好的容错性和性能优势，因此成为了大数据处理的热门选择。

在Spark中，机器学习库MLlib提供了丰富的机器学习算法和工具，方便用户进行各种机器学习任务的实现。

MLlib支持常见的机器学习任务，例如分类、回归、聚类和推荐等。

用户可以利用MLlib提供的算法和工具，快速构建并训练自己的模型。

在机器学习应用中，数据预处理是一个必不可少的步骤。

Spark提供了丰富的数据处理函数和工具，可以帮助用户对数据进行清洗、转换和特征提取等操作。

例如，用户可以使用Spark的数据清洗函数对数据中的缺失值进行处理，使用特征提取算法将原始数据转换为可用于训练的特征向量。

除了数据预处理，特征工程也是机器学习应用中非常重要的一步。

在特征工程中，用户可以利用Spark强大的数据处理能力，进行特征选择、特征降维和特征构造等操作。

Spark提供了多种特征选择算法，例如互信息、卡方检验和递归特征消除等，帮助用户选择最具有代表性和区分度的特征。

此外，Spark还提供了主成分分析（PCA）和线性判别分析（LDA）等降维算法，方便用户将高维数据转换为低维空间。

同时，用户还可以根据自己的业务需求，利用Spark的工具进行特征构造，例如利用文本数据进行词袋模型的构建，或者利用时间序列数据进行滑动窗口的构建。

在模型训练阶段，Spark提供了丰富的机器学习算法，方便用户选择适合自己问题的算法。

基于Spark的MLlibExample 1: K-means（一）算法简介K-means算法是典型的基于距离的聚类算法，采用距离作为相似性的评价指标，即认为两个对象的距离越近，其相似度就越大。

该算法认为簇是由距离靠近的对象组成的，因此把得到紧凑且独立的簇作为最终目标。

k个初始类聚类中心点的选取对聚类结果具有较大的影响，因为在该算法第一步中是随机的选取任意k个对象作为初始聚类的中心，初始地代表一个簇。

该算法在每次迭代中对数据集中剩余的每个对象，根据其与各个簇中心的距离将每个对象重新赋给最近的簇。

当考察完所有数据对象后，一次迭代运算完成，新的聚类中心被计算出来。

如果在一次迭代前后，评价指标的值没有发生变化，说明算法已经收敛。

（二）算法步骤算法过程如下：Step 1: 从数据集的N个元素中随机选取K个元素作为质心；Step 2: 对剩余的每个元素测量其到每个质心的距离，并把它归到最近的质心的类；Step 3: 重新计算已经得到的各个类的质心；Step 4: 迭代2～3步直至新的质心与原质心相等或小于指定阈值，算法结束。

（三）算法举例举个简单的例子，假如现在有n个数据形成的数据集data[n]，要对这n个元素进行k-means聚类，那么可按以下流程去做：输入：k,data[n];Step 1: 选择k个初始中心点，例如c[0]=data[0],…c[k−1]=data[k−1]；Step 2: 对于data[0]….data[n]，分别与c[0]…c[k−1]进行比较，假定与c[i]差值最少，就标记为i；Step 3: 对于所有标记为i的点，重新计算它们的中心c[i]= {所有标记为i的data[j]之和}/标记为i的个数；Step 4: 重复step 2 和step 3，直到所有c[i]值的变化小于给定阈值。

（四）基于Spark的算法实现并行化说明：可以找到以上代码中标记为Trains the k-means model的那行代码，可以看到这里调用了一个命名为KMeans的包，我们去找到这个包，里面做的就是数据的切割和分发处理，第二个标黄的reduceByKey要做的事情就是把每一块处理完的结果进行合并汇总。

SparkMLib完整基础⼊门教程Spark MLib在Spark下进⾏机器学习，必然⽆法离开其提供的MLlib框架，所以接下来我们将以本框架为基础进⾏实际的讲解。

⾸先我们需要了解其中最基本的结构类型，即转换器、估计器、评估器和流⽔线。

graph LR A[转换器] --> B(估计器) B --> C(评估器) C --> D[模型]⾸先欢迎⼤家Start本⼈关于机器学习的，不仅仅包含了Spark ML还包括python下的sklearn等主流库。

⼀、基础使⽤接下来我们将以⼀个简单的例⼦为基础整体介绍在Spark下进⾏机器学习的使⽤⽅式，便于读者⼤体熟悉完整的流程节点。

当然在这其中对于部分不了解的情况下可以等在后续详细学习的过程中进⾏补充即可。

1. 特征⼯程这部分相关知识可以参考本⼈编写的的开源教程，其中对该部分进⾏详细的说明，下⾯我们将就框架提供的RFormula进⾏具体的实战操作（这⾥熟悉R语⾔的可能对此⽐较熟悉，本⾝就是借鉴了R语⾔，但是仅实现了其中的⼀个⼦集），对于我们需要进⾏特征化的数据⾸先我们需要定义对应的线性模型公式，具体如下。

Dataset<Row> df = session.read().json("sparkdemo/data/simple-ml");RFormula supervised = new RFormula().setFormula("lab ~ . + color: value1 + color: value2");当然仅仅通过上述的⽅式还不能实现对数据的特征化，我们还需要通过数据对其进⾏训练，从⽽得到我们所需的转换器，为此我们需要使⽤其中的fit⽅法进⾏转换。

RFormulaModel model = supervised.fit(df);完成转换器的训练后我们就可以利⽤其进⾏实际的转换操作，从⽽⽣成特征features与标签label列，当然读者也可以通过supervised.setLabelCol设置标签列名，supervised.setFeaturesCol设置特征列名。

Spark基本概念及⼊门sparkspark背景什么是sparkSpark是⼀种快速、通⽤、可扩展的⼤数据分析引擎，2009年诞⽣于加州⼤学伯克利分校AMPLab，2010年开源，2013年6⽉成为Apache孵化项⽬，2014年2⽉成为Apache顶级项⽬。

⽬前，Spark⽣态系统已经发展成为⼀个包含多个⼦项⽬的集合，其中包含SparkSQL、Spark Streaming、GraphX、MLlib等⼦项⽬，Spark是基于内存计算的⼤数据并⾏计算框架。

Spark基于内存计算，提⾼了在⼤数据环境下数据处理的实时性，同时保证了⾼容错性和⾼可伸缩性，允许⽤户将Spark部署在⼤量廉价硬件之上，形成集群。

Spark与HadoopSpark是⼀个计算框架,⽽Hadoop中包含计算框架MapReduce和分布式⽂件系统HDFS,Hadoop更⼴泛地说还包括在其⽣态系统上的其他系统.为什么使⽤Spark?Hadoop的MapReduce计算模型存在问题:Hadoop的MapReduce的核⼼是Shuffle(洗牌).在整个Shuffle的过程中,⾄少产⽣6次I/O流.基于MapReduce计算引擎通常会将结果输出到次盘上,进⾏存储和容错.另外,当⼀些查询(如:hive)翻译到MapReduce任务是,往往会产⽣多个Stage,⽽这些Stage有依赖底层⽂件系统来存储每⼀个Stage的输出结果,⽽I/O的效率往往较低,从⽽影响MapReduce的运⾏速度.Spark的特点: 快, 易⽤, 通⽤,兼容性快：与Hadoop的MapReduce相⽐，Spark基于内存的运算要快100倍以上，基于硬盘的运算也要快10倍以上。

Spark实现了⾼效的DAG执⾏引擎，可以通过基于内存来⾼效处理数据流。

易⽤：Spark⽀持Java、Python和Scala的API，还⽀持超过80种⾼级算法，使⽤户可以快速构建不同的应⽤。

⽽且Spark⽀持交互式的Python和Scala的shell，可以⾮常⽅便地在这些shell中使⽤Spark集群来验证解决问题的⽅法。

Spark基础知识详解Apache Spark是⼀种快速通⽤的集群计算系统。

它提供Java，Scala，和R中的⾼级API，以及⽀持通⽤执⾏图的优化引擎。

它还⽀持⼀组丰富的⾼级⼯具，包括⽤于SQL和结构化数据处理的Spark SQL，⽤于机器学习的MLlib，⽤于图形处理的GraphX和Spark Streaming。

Spark优点：减少磁盘I/O：随着实时⼤数据应⽤越来越多，Hadoop作为离线的⾼吞吐、低响应框架已不能满⾜这类需求。

HadoopMapReduce的map端将中间输出和结果存储在磁盘中，reduce端⼜需要从磁盘读写中间结果，势必造成磁盘IO成为瓶颈。

Spark允许将map端的中间输出和结果存储在内存中，reduce端在拉取中间结果时避免了⼤量的磁盘I/O。

Hadoop Yarn中的ApplicationMaster申请到Container后，具体的任务需要利⽤NodeManager从HDFS的不同节点下载任务所需的资源（如Jar包），这也增加了磁盘I/O。

Spark将应⽤程序上传的资源⽂件缓冲到Driver本地⽂件服务的内存中，当Executor执⾏任务时直接从Driver的内存中读取，也节省了⼤量的磁盘I/O。

增加并⾏度：由于将中间结果写到磁盘与从磁盘读取中间结果属于不同的环节，Hadoop将它们简单的通过串⾏执⾏衔接起来。

Spark把不同的环节抽象为Stage，允许多个Stage 既可以串⾏执⾏，⼜可以并⾏执⾏。

避免重新计算：当Stage中某个分区的Task执⾏失败后，会重新对此Stage调度，但在重新调度的时候会过滤已经执⾏成功的分区任务，所以不会造成重复计算和资源浪费。

可选的Shuffle排序：HadoopMapReduce在Shuffle之前有着固定的排序操作，⽽Spark则可以根据不同场景选择在map端排序或者reduce端排序。

灵活的内存管理策略：Spark将内存分为堆上的存储内存、堆外的存储内存、堆上的执⾏内存、堆外的执⾏内存4个部分。

1、机器学习概念1.1机器学习的定义在维基百科上对机器学习提出以下几种定义：●“机器学习是一门人工智能的科学，该领域的主要研究对象是人工智能，特别是如何在经验学习中改善具体算法的性能”。

●“机器学习是对能通过经验自动改进的计算机算法的研究”。

●“机器学习是用数据或以往的经验，以此优化计算机程序的性能标准。

”一种经常引用的英文定义是：A computer program is said to learn fromexperience E with respect to some class of tasks T and performance measure P, if its performance at tasks in T, as measured by P,improves with experience E。

可以看出机器学习强调三个关键词：算法、经验、性能，其处理过程如下图所示。

上图表明机器学习是数据通过算法构建出模型并对模型进行评估，评估的性能如果达到要求就拿这个模型来测试其他的数据，如果达不到要求就要调整算法来重新建立模型，再次进行评估，如此循环往复，最终获得满意的经验来处理其他的数据。

1.2机器学习的分类1.2.1 监督学习监督是从给定的训练数据集中学习一个函数（模型），当新的数据到来时，可以根据这个函数（模型）预测结果。

监督学习的训练集要求包括输入和输出，也可以说是特征和目标。

训练集中的目标是由人标注（标量）的。

在监督式学习下，输入数据被称为“训练数据”，每组训练数据有一个明确的标识或结果，如对防垃圾邮件系统中“垃圾邮件”、“非垃圾邮件”，对手写数字识别中的“1”、“2”、“3”等。

在建立预测模型时，监督式学习建立一个学习过程，将预测结果与“训练数据”的实际结果进行比较，不断调整预测模型，直到模型的预测结果达到一个预期的准确率。

常见的监督学习算法包括回归分析和统计分类：●二元分类是机器学习要解决的基本问题，将测试数据分成两个类，如垃圾邮件的判别、房贷是否允许等问题的判断。

Sparkmllib随机森林算法的简单应⽤（附代码）此前⽤⾃⼰实现的随机森林算法，应⽤在titanic⽣还者预测的数据集上。

事实上，有很多开源的算法包供我们使⽤。

⽆论是本地的机器学习算法包sklearn 还是分布式的spark mllib，都是⾮常不错的选择。

Spark是⽬前⽐较流⾏的分布式计算解决⽅案，同时⽀持集群模式和本地单机模式。

由于其通过scala语⾔开发，原⽣⽀持scala，同时由于python在科学计算等领域的⼴泛应⽤，Spark也提供了python的接⼝。

Spark的常⽤操作详见官⽅⽂档：/docs/latest/programming-guide.html在终端下⾯键⼊如下命令，切换到spark的⽬录，进⼊相应的环境：cd $SPARK_HOMEcd ./bin./pyspark可以看到，出现了python 的版本号以及spark的logo此时，仍然是输⼊⼀句，运⾏⼀句并输出。

可以事先编辑好脚本保存为filename然后：./spark-submit filename下⾯给出详细的代码：[python]1. import pandas as pd2. import numpy as np3. from pyspark.mllib.regression import LabeledPoint4. from pyspark.mllib.tree import RandomForest5.6.7. #将类别数量⼤于2的类别型变量进⾏重新编码，并把数据集变成labeledPoint格式8. #df=pd.read_csv('/home/kim/t.txt',index_col=0)9. #for col in ['Pclass','embrk']:10. # values=df[col].drop_duplicates()11. # for v in values:12. # col_name=col+str(v)13. # df[col_name]=(df[col]==v)14. # df[col_name]=df[col_name].apply(lambda x:int(x))15. #df=df.drop(['Pclass','embrk'],axis=1)16. #df.to_csv('train_data')17.18. #读⼊数据集变成弹性分布式数据集RDD ，由于是有监督学习，需要转换为模型输⼊的格式LabeledPoint19. rdd=pyspark.SparkContext.textFile('/home/kim/train')20. train=rdd.map(lambda x:x.split(',')[1])21. train=train.map(lambda line:LabeledPoint(line[1],line[2:]))22.23. #模型训练24. model=RandomForest.trainClassifier\25. (train, numClasses=2, categoricalFeaturesInfo={},numTrees=1000,\26. featureSubsetStrategy="auto",impurity='gini', maxDepth=4, maxBins=32)27.28. #包含LabeledPoint对象的RDD，应⽤features⽅法返回其输⼊变量的值,label⽅法返回其真实类别29. data_p=train.map(lambda lp:lp.features)30. v=train.map(lambda lp:bel)31. prediction=model.predict(data_p)32. vp=v.zip(prediction)33.34. #最后输出模型在训练集上的正确率35. MSE=vp.map(lambda x:abs(x[0]-x[1]).sum())/vp.count()36. print("MEAN SQURE ERROR: "+str(MSE))import pandas as pdimport numpy as npfrom pyspark.mllib.regression import LabeledPointfrom pyspark.mllib.tree import RandomForest#将类别数量⼤于2的类别型变量进⾏重新编码，并把数据集变成labeledPoint格式#df=pd.read_csv('/home/kim/t.txt',index_col=0)#for col in ['Pclass','embrk']:# values=df[col].drop_duplicates()# for v in values:# col_name=col+str(v)# df[col_name]=(df[col]==v)# df[col_name]=df[col_name].apply(lambda x:int(x))#df=df.drop(['Pclass','embrk'],axis=1)#df.to_csv('train_data')#读⼊数据集变成弹性分布式数据集RDD ，由于是有监督学习，需要转换为模型输⼊的格式LabeledPoint rdd=pyspark.SparkContext.textFile('/home/kim/train')train=rdd.map(lambda x:x.split(',')[1])train=train.map(lambda line:LabeledPoint(line[1],line[2:]))#模型训练model=RandomForest.trainClassifier\(train, numClasses=2, categoricalFeaturesInfo={},numTrees=1000,\featureSubsetStrategy="auto",impurity='gini', maxDepth=4, maxBins=32)#包含LabeledPoint对象的RDD，应⽤features⽅法返回其输⼊变量的值,label⽅法返回其真实类别data_p=train.map(lambda lp:lp.features)v=train.map(lambda lp:bel)prediction=model.predict(data_p)vp=v.zip(prediction)#最后输出模型在训练集上的正确率MSE=vp.map(lambda x:abs(x[0]-x[1]).sum())/vp.count()print("MEAN SQURE ERROR: "+str(MSE))后⾯可以多加测试，例如：使⽤更⼤规模的数据集；将数据集划分为训练集测试集，在训练集上建模在测试集上评估模型性能；使⽤mllib⾥⾯的其他算法并⽐较效果，等等欢迎⼤家与我交流！。