在这篇文章中，我们Spark的其它机器学习API，名为Spark ML，如果要用数据流水线来开发大数据应用程序的话，这个是推荐的解决方案。关键点：

• 了解机器学习数据流水线有关内容。
• 怎么用Apache Spark机器学习包来实现机器学习数据流水线。
• 数据价值链处理的步骤。
• Spark机器学习流水线模块和API。
• 文字分类和广告检测用例。

Spark ML(spark.ml)包提供了构建在DataFrame之上的机器学习API，它已经成了Spark SQL库的核心部分。这个包可以用于开发和管理机器学习流水线。它也可以提供特征抽取器、转换器、选择器，并支持分类、汇聚和分簇等机器学习技术。这些全都对开发机器学习解决方案至关重要。

在这里我们看看如何使用Apache Spark来做探索式数据分析(Exploratory Data Analysis)、开发机器学习流水线，并使用Spark ML包中提供的API和算法。

因为支持构建机器学习数据流水线，Apache Spark框架现在已经成了一个非常不错的选择，可以用于构建一个全面的用例，包括ETL、指量分析、实时流分析、机器学习、图处理和可视化等。

机器学习数据流水线

机器学习流水线可以用于创建、调节和检验机器学习工作流程序等。机器学习流水线可以帮助我们更加专注于项目中的大数据需求和机器学习任务等，而不是把时间和精力花在基础设施和分布式计算领域上。它也可以在处理机器学习问题时帮助我们，在探索阶段我们要开发迭代式功能和组合模型。

机器学习工作流通常需要包括一系列的处理和学习阶段。机器学习数据流水线常被描述为一种阶段的序列，每个阶段或者是一个转换器模块，或者是个估计器模块。这些阶段会按顺序执行，输入数据在流水线中流经每个阶段时会被处理和转换。

机器学习开发框架要支持分布式计算，并作为组装流水线模块的工具。还有一些其它的构建数据流水线的需求，包括容错、资源管理、可扩展性和可维护性等。

在真实项目中，机器学习工作流解决方案也包括模型导入导出工具、交叉验证来选择参数、为多个数据源积累数据等。它们也提供了一些像功能抽取、选择和统计等的数据工具。这些框架支持机器学习流水线持久化来保存和导入机器学习模型和流水线，以备将来使用。

机器学习工作流的概念和工作流处理器的组合已经在多种不同系统中越来越受欢迎。象scikit-learn和GraphLab等大数据处理框架也使用流水线的概念来构建系统。

一个典型的数据价值链流程包括如下步骤：

• 发现
• 注入
• 处理
• 保存
• 整合
• 分析
• 展示

机器学习数据流水线所用的方法都是类似的。下图展示了在机器学习流水线处理中涉及到的不同步骤。

表一：机器学习流水线处理步骤

这些步骤也可以用下面的图一表示。 

图一：机器学习数据流水线处理流图

接下来让我们一起看看每个步骤的细节。

数据注入：我们收集起来供给机器学习流水线应用程序的数据可以来自于多种数据源，数据规模也是从几百GB到几TB都可以。而且，大数据应用程序还有一个特征，就是注入不同格式的数据。

数据清洗：数据清洗这一步在整个数据分析流水线中是第一步，也是至关重要的一步，也可以叫做数据清理或数据转换，这一步主要是要把输入数据变成结构化的，以方便后续的数据处理和预测性分析。依进入到系统中的数据质量不同，总处理时间的60%-70%会被花在数据清洗上，把数据转成合适的格式，这样才能把机器学习模型应用到数据上。

数据总会有各种各样的质量问题，比如数据不完整，或者数据项不正确或不合法等。数据清洗过程通常会使用各种不同的方法，包括定制转换器等，用流水线中的定制的转换器去执行数据清洗动作。

稀疏或粗粒数据是数据分析中的另一个挑战。在这方面总会发生许多极端案例，所以我们要用上面讲到的数据清洗技术来保证输入到数据流水线中的数据必须是高质量的。

伴随着我们对问题的深入理解，每一次的连续尝试和不断地更新模型，数据清洗也通常是个迭代的过程。象Trifacta、OpenRefine或ActiveClean等数据转换工具都可以用来完成数据清洗任务。

特征抽取：在特征抽取(有时候也叫特征工程)这一步，我们会用特征哈希(Hashing Term Frequency)和Word2Vec等技术来从原始数据中抽取具体的功能。这一步的输出结果常常也包括一个汇编模块，会一起传入下一个步骤进行处理。

模型训练：机器学习模型训练包括提供一个算法，并提供一些训练数据让模型可以学习。学习算法会从训练数据中发现模式，并生成输出模型。

模型验证：这一步包评估和调整机器学习模型，以衡量用它来做预测的有效性。如这篇文章所说，对于二进制分类模型评估指标可以用接收者操作特征(Receiver Operating Characteristic，ROC)曲线。ROC曲线可以表现一个二进制分类器系统的性能。创建它的方法是在不同的阈值设置下描绘真阳性率(True Positive Rate，TPR)和假阳性率(False Positive Rate，FPR)之间的对应关系。

模型选择：模型选择指让转换器和估计器用数据去选择参数。这在机器学习流水线处理过程中也是关键的一步。ParamGridBuilder和CrossValidator等类都提供了API来选择机器学习模型。

模型部署：一旦选好了正确的模型，我们就可以开始部署，输入新数据并得到预测性的分析结果。我们也可以把机器学习模型部署成网页服务。

Spark机器学习

机器学习流水线API是在Apache Spark框架1.2版中引入的。它给开发者们提供了API来创建并执行复杂的机器学习工作流。流水线API的目标是通过为不同机器学习概念提供标准化的API，来让用户可以快速并轻松地组建并配置可行的分布式机器学习流水线。流水线API包含在org.apache.spark.ml包中。

Spark ML也有助于把多种机器学习算法组合到一条流水线中。

Spark机器学习API被分成了两个包，分别是spark.mllib和spark.ml。其中spark.ml包包括了基于RDD构建的原始API。而spark.ml包则提供了构建于DataFrame之上的高级API，用于构建机器学习流水线。

基于RDD的MLlib库API现在处于维护模式。

如下面图二所示，Spark ML是Apache Spark生态系统中的一个非常重要的大数据分析库。

图二：包括了Spark ML的Spark生态系统

机器学习流水线模块

机器学习数据流水线包括了完成数据分析任务所需要的多个模块。数据流水线的关键模块被列在了下面：

• 数据集
• 流水线
• 流水线的阶段
• 转换器
• 估计器
• 评估器
• 参数(和参数地图)

接下来我们简单看看这些模块可以怎么对应到整体的步骤中。

数据集：在机器学习流水线中是使用DataFrame来表现数据集的。它也允许按有名字的字段保存结构化数据。这些字段可以用于保存文字、功能向量、真实标签和预测。

流水线：机器学习工作流被建模为流水线，这包括了一系列的阶段。每个阶段都对输入数据进行处理，为下一个阶段产生输出数据。一个流水线把多个转换器和估计器串连起来，描述一个机器学习工作流。

流水线的阶段：我们定义两种阶段，转换器和估计器。

转换器：算法可以把一个DataFrame转换成另一个DataFrame。比如，机器学习模型就是一个转换器，用于把一个有特征的DataFrame转换成一个有预测信息的DataFrame。

转换器会把一个DataFrame转成另一个DataFrame，同时为它加入新的特征。比如在Spark ML包中，OneHotEncoder就会把一个有标签索引的字段转换成一个有向量特征的字段。每个转换器都有一个transform()函数，被调用时就会把一个DataFrame转换成另一个。

估计器：估计器就是一种机器学习算法，会从你提供的数据中进行学习。估计器的输入是一个DataFrame，输出就是一个转换器。估计器用于训练模型，它生成转换器。比如，逻辑回归估计器就会产生逻辑回归转换器。另一个例子是把K-Means做为估计器，它接受训练数据，生成K-Means模型，就是一个转换器。

参数：机器学习模块会使用通用的API来描述参数。参数的例子之一就是模型要使用的最大迭代次数。

下图展示的是一个用作文字分类的数据流水线的各个模块。

图三：使用Spark ML的数据流水线

用例

机器学习流水线的用例之一就是文字分类。这种用例通常包括如下步骤：

• 清洗文字数据
• 将数据转化成特征向量，并且
• 训练分类模型

在文字分类中，在进行分类模型(类似SVM)的训练之前，会进行n-gram抽象和TF-IDF特征权重等数据预处理。

另一个机器学习流水线用例就是在这篇文章中描述的图像分类。

还有很多种其它机器学习用例，包括欺诈检测(使用分类模型，这也是监督式学习的一部分)，用户分区(聚簇模型，这也是非监督式学习的一部分)。

TF-IDF

词频-逆向文档频率(Term Frequency - Inverse Document Frequency，TF-IDF)是一种在给定样本集合内评估一个词的重要程度的静态评估方法。这是一种信息获取算法，用于在一个文档集合内给一个词的重要性打分。

TF：如果一个词在一份文档中反复出现，那这个词就比较重要。具体计算方法为：

TF = (# of times word X appears in a document) / (Total # of

words in the document)

IDF：但如果一个词在多份文档中都频繁出现(比如the，and，of等)，那就说明这个词没有什么实际意义，因此就要降低它的评分。

示例程序

下面我们看个示例程序，了解一下Spark ML包可以怎样用在大数据处理系统中。我们会开发一个文档分类程序，用于区别程序输入数据中的广告内容。测试用的输入数据集包括文档、电子邮件或其它任何从外部系统中收到的可能包含广告的内容。

我们将使用在Strata Hadoop World Conference研讨会上讨论的“用Spark构建机器学习应用”的广告检测示例来构建我们的示例程序。

用例

这个用例会对发送到我们的系统中的各种不同消息进行分析。有些消息里面是含有广告信息的，但有些消息里面没有。我们的目标就是要用Spark ML API找出那些包含了广告的消息。

算法

我们将使用机器学习中的逻辑回归算法。逻辑回归是一种回归分析模型，可以基于一个或多个独立变量来预测得到是或非的可能结果。

详细的解决方案

接下来咱们看看这个Spark ML示例程序的细节，以及运行步骤。

数据注入：我们会把包含广告的数据(文本文件)和不包含广告的数据都导入。

数据清洗：在示例程序中，我们不做任何特别的数据清洗操作。我们只是把所有的数据都汇聚到一个DataFrame对象中。

我们随机地从训练数据和测试数据中选择一些数据，创建一个数组对象。在这个例子中我们的选择是70%的训练数据，和30%的测试数据。

在后续的流水线操作中我们分别用这两个数据对象来训练模型和做预测。

我们的机器学习数据流水线包括四步：

• Tokenizer
• HashingTF
• IDF
• LR

创建一个流水线对象，并且在流水线中设置上面的各个阶段。然后我们就可以按照例子，基于训练数据来创建一个逻辑回归模型。

现在，我们再使用测试数据(新数据集)来用模型做预测。

下面图四中展示了例子程序的架构图。

图4：数据分类程序架构图

技术

在实现机器学习流水线解决方案时我们用到了下面的技术。

表二：在机器学习例子中用到的技术和工具

Spark ML程序

根据研讨会上的例子而写成的机器学习代码是用Scala编程语言写的，我们可以直接使用Spark Shell控制台来运行这个程序。

广告检测Scala代码片段：

第一步：创建一个定制的类，用来存储广告内容的细节。

1. case class SpamDocument(file: String, text: String, label: 
2.  
3. Double)  

第二步：初始化SQLContext，并通过隐式转换方法来把Scala对象转换成DataFrame。然后从存放着输入文件的指定目录导入数据集，结果会返回RDD对象。然后由这两个数据集的RDD对象创建DataFrame对象。

1. val sqlContext = new SQLContext(sc) 
2. import sqlContext.implicits._ 
3.  
4. // 
5. // Load the data files with spam 
6. // 
7. val rddSData = sc.wholeTextFiles("SPAM_DATA_FILE_DIR", 1) 
8. val dfSData = rddSData.map(d => SpamDocument(d._1, d._2,1)).toDF() 
9. dfSData.show() 
10.  
11. // 
12. // Load the data files with no spam 
13. // 
14. val rddNSData = sc.wholeTextFiles("NO_SPAM_DATA_FILE_DIR", 
15. 1) 
16. val dfNSData = rddNSData.map(d => SpamDocument(d._1,d._2, 0)).toDF() 
17. dfNSData.show()  

第三步：现在，把数据集汇聚起来，然后根据70%和30%的比例来把整份数据拆分成训练数据和测试数据。

1. // 
2. // Aggregate both data frames 
3. // 
4. val dfAllData = dfSData.unionAll(dfNSData) 
5. dfAllData.show() 
6.  
7. // 
8. // Split the data into 70% training data and 30% test data 
9. // 
10. val Array(trainingData, testData) = 
11. dfAllData.randomSplit(Array(0.7, 0.3))  

第四步：现在可以配置机器学习数据流水线了，要创建我们在文章前面部分讨论到的几个部分：Tokenizer、HashingTF和IDF。然后再用训练数据创建回归模型，在这个例子中是逻辑回归。

1. // 
2. // Configure the ML data pipeline 
3. // 
4.  
5. // 
6. // Create the Tokenizer step 
7. // 
8. val tokenizer = new Tokenizer() 
9.   .setInputCol("text") 
10.   .setOutputCol("words") 
11.  
12. // 
13. // Create the TF and IDF steps 
14. // 
15. val hashingTF = new HashingTF() 
16.   .setInputCol(tokenizer.getOutputCol) 
17.   .setOutputCol("rawFeatures") 
18.  
19. val idf = new 
20. IDF().setInputCol("rawFeatures").setOutputCol("features") 
21.  
22. // 
23. // Create the Logistic Regression step 
24. // 
25. val lr = new LogisticRegression() 
26.   .setMaxIter(5) 
27. lr.setLabelCol("label") 
28. lr.setFeaturesCol("features") 
29.  
30. // 
31. // Create the pipeline 
32. // 
33. val pipeline = new Pipeline() 
34.   .setStages(Array(tokenizer, hashingTF, idf, lr)) 
35.  
36. val lrModel = pipeline.fit(trainingData) 
37. println(lrModel.toString())  

第五步：最后，我们调用逻辑回归模型中的转换方法来用测试数据做预测。

1. // 
2. // Make predictions. 
3. // 
4. val predictions = lrModel.transform(testData) 
5.  
6. // 
7. // Display prediction results 
8. // 
9. predictions.select("file", "text", "label", "features", "prediction").show(300)  

结论

Spark机器学习库是Apache Spark框架中最重要的库之一。它用于实现数据流水线。在这篇文章中，我们了解了如何使用Spark ML包的API以及用它来实现一个文本分类用例。

接下来的内容

图数据模型是关于在数据模型中不同的实体之间的连接和关系的。图数据处理技术最近受到了很多关注，因为可以用它来解决许多问题，包括欺诈检测和开发推荐引擎等。

Spark框架提供了一个库，专门用于图数据分析。我们在这个系列的文章中，接下来会了解这个名为Spark GraphX的库。我们会用Spark GraphX来开发一个示例程序，用于图数据处理和分析。

本文作者：足下

来源：51CTO