Spark MLlib学习——综述和Pipeline

2019年6月8日 176次阅读来源: shohokuooo

MLlib是Spark的机器学习lib，目的是让机器学习的实践变得更加简单，总的来说它提供了以下几种工具：

ML算法：分类、回归、聚类和协同过滤等常用学习算法
特征工程：特征提取、转换、降维和选择
Pipelines：创建、评价和调优的ML Pipelines工具
存储：保存和加载算法、模型和Pipeline
实用工具：线性代数、统计、数据处理等

需要添加的Maven依赖：

《Spark MLlib学习——综述和Pipeline》

DataFrame的API是最主要的API

从Spark 2.0开始，RDD接口的API进入到了维护状态，所以现在机器学习的最主要的API是DataFrame接口的API。

这意味着：

1、MLlib仍然支持修复bug的RDD接口的API
2、MLlib不会向RDD的API加入新的特性了
3、在接下来的2.x版本中MLlib将持续添加DataFrame接口的API的特性来达到和RDD API相等的程度。
4、在达到相等的程度后RDD接口的API将会被抛弃。（粗略估计在2.2版本）
5、预计在3.0版本上彻底移除RDD接口的API

为什么MLlib转去使用DataFrame接口的API

1、DataFrame接口的API对用户更加友好。包括使用Spark的Datasources，SQL/DataFrame查询语句，Tungsten和Catalyst执行优化器，以及跨语言统一的API接口
2、DataFrame帮助更好使用ML Pipeline，尤其是Transformation特性。

依赖包

MLlib使用了线性代数包Breeze，这个包依赖来netlib-java优化数值计算。如果运行时native libraries不可用，你将看到一个警告提示信息，然后使用纯虚拟机进行处理。

因为licence的缘故，我们默认不使用netlib-java的native proxies。配置netlib-java/Breeze来使用系统的优化库，包括在你的工程中加入com.github.fommil.netlib:all:1.1.2（或者build Spark的时候带上-Pnetlib-lgpl）作为依赖，请阅读netlib-java官方文档来获得安装说明。（目前运行后面example代码的时候还是出现warn显示Native BLAS的加载有问题。这个问题真是有些头大。。。）

基于Python语言使用MLlib，需要安装1.4以上版本Numpy。

ML Pipelines

下面开始介绍ML Pipelines。虽然 MLlib 已经足够简单易用，但是如果目标数据集结构复杂需要多次处理，或者是对新数据进行预测的时候需要结合多个已经训练好的单个模型进行综合预测 (集成学习的思想)，那么使用 MLlib 将会让程序结构复杂，难于理解和实现。值得庆幸的是，在 Spark 的生态系统里，一个可以用于构建复杂机器学习工作流应用的新库已经出现了，它就是 Spark 1.2 版本之后引入的 ML Pipeline。

====================

Pipelines中的主要概念

基于MLlib提供的机器学习算法的标准API，我们可以非常简单的组合多种不同的算法成为一个pipeline或者workflow。下面将会介绍Pipeline API中的几个关键概念，其中这些概念主要是受项目scikit-learn的启发。

DataFrame:ML API使用这个来自Spark SQL的概念作为ML dataset，可以保存多种数据类型。比如：使用不同的列存储文本、特征向量、真实标签和预测结果。
Transformer:这是个是指一个算法将一个DataFrame transform成另一个DataFrame。也就是训练好的模型。比如：一个ML模型就是一个Transformer能够将一个特征数据的DataFrame转成预测结果的DataFrame。
Estimator:是指一个操作DataFrame产生Transformer的算法。比如：一个学习算法就是一个Estimator，可以在一个DataFrame上训练得到一个模型。
Pipeline:一个Pipeline链是将多个Transformer和Estimator组合在一起组成一个ML workflow。
Parameter:所有的Transformer和Estimator共享一个公共的说明参数的API。

Pipeline组件

Transformer

Transformer是一个抽象类，包括特征转换以及学习后的模型两种情况。不管是哪一种都需要实现方法transform()，用以将一个DataFrame转换成另一个，通常都是添加一个或多个列。

比如：

特征转换是：读取一个列（比如：text）然后映射为一个新列（比如：特征向量），输出的新DataFrame就是将新映射出来的列加到原来的DataFrame中。
学习后的模型：读取一个DataFrame（包含特征向量），对于每个特征向量预测label，最后输出携带预测结果的新的DataFrame。

Estimator

这个是一个学习算法的抽象，可以对数据fit或者train。一个Estimator需要实现方法fit()，这个方法接收一个DataFrame并输出一个模型也就是Transformer。比如：LogisticRegression就是一个Estimator，通过调用fit()训练得到LogisticRegressionModel，这个输出模型也就是Transformer。

===================

Pipeline

在机器学习中，通常有一系列的算法在数据中处理和学习。比如：一个简单的数据文本处理流程大概分为下面几个阶段：

将文本分割成words
将words转化为数值型的特征向量
通过特征向量和标签学习得到一个预测模型

MLlib将上述workflow定义为Pipeline，它包含了一连串的PipelineStage（Transformer和Estimator）按照特定的顺序运行。

如何工作

Pipeline定义了一连串的Stage，每个Stage不是Transformer就是Estimator，这些Stage按照顺序执行。

对于一个简单文本的处理流程，下图展示了Pipeline的training time

《Spark MLlib学习——综述和Pipeline》从训练集中学习（Estimator）过程

在上图中第一行包括三个Stage，蓝色的前两个Stage（Tokenizer和HashingTF）是Transformer类型的，最后红色的LogisticRegression是Estimator类型的。第二行中的每个圆柱体都表示DataFrame。经过Tokenizer.transform()、HashingTF.transform()以及LogisticRegression.fit()之后，产生LogisticRegressionModel。

接下来就是调用LogisticRegressionModel’stransform()如下图所示，被称为test time

《Spark MLlib学习——综述和Pipeline》在测试集上预测的过程

细节

DAGPipelines: Pipeline的Stage被规定为有序数组。在上面简单用例中可以看到是线性的，也就是后一个Stage用到的数据是前一个Stage的输出。当然也可以创建非线性的Pipeline只要数据流图是DAG的。

Runtime checking:因为DataFrame支持多种数据类型，所以不能可能在编译时进行检查，只能在运行时通过DataFrame的schema进行检查。

Unique Pipeline stages: Pipeline中的每一个Stage都应该是唯一性的实例。比如同一个实例myHashingTF不应该插入到Pipeline的中两次，也就是说Pipeline的每个Stage的ID都必须是唯一的。然而同一个类型的两个不同的实例可以插入一个Pipeline中，并创建不同的ID。

Parameters

MLlib的Estimator和Transformer使用统一的API来管理参数。

ParamMap是一个(parameter, value)的集合。

有两种方法对算法传入参数：

1、通过实例来设置参数。比如：lr是LogisticRegression的一个实例，那么lr.setMaxIter(10)就是让lr在调用fit()方法的时候使用最大10次迭代的参数。
2、通过ParamMap对fit()和transform()设置参数。如果在方法fit()或者transform()中传入ParamMap的对象那么就会覆盖之前的参数设置。

举例如果有两个LogisticRegression类型的实例lr1和lr2，我们可以使用ParamMap这样对参数赋值ParamMap(lr1.maxIter -> 10, lr2.maxIter -> 20)。具体Java代码实例看下面