[Spark2.0]ML piplines管道模式

Posted 2023-03-01 yhao浩

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了[Spark2.0]ML piplines管道模式相关的知识，希望对你有一定的参考价值。

在本部分，我们将介绍ML Pipline的概念。ML Pipline提供了一整套建立在DataFrame上的高级API，它能帮助用户创建和调优实际机器学习管道。

Pipline的主要思想

Mllib标准化了机器学习算法的API，使得将多个算法融合到一个简单的管道或工作流更为简单。本部分将覆盖Pipline API的关键思想，这里的pipline概念是受scikit-learn项目启发而来。

DataFrame：此ML API使用从Spark SQL而来的DataFrame作为ML的数据集。DataFrame可以包含各种数据类型。例如，DataFrame可以有不同的列用于存储文本、特征向量、真实标签以及预测值。
Transformer：Transformer是一种能够把一个DataFrame转换为另一个DataFrame的算法。例如，一个ML模型就是一个可以将一个包含特征的DataFrame转换为包含预测值的DataFrame的Transformer。
Estimator：Estimator是能够使用在DataFrame上用以产生一个Transformer的算法。例如，一个学习算法就是一个能够在DataFrame上进行训练并产生模型的Estimator。
Pipline：Pipline可以将多个Transformer和Estimator链接到一起形成一个ML工作流。
Parameter：所有的Transformer和Estimator共享一个通用的指定参数的API。

DataFrame

机器学习可以应用于各种各样的数据类型，如向量，文本，图片和结构化数据。此API采用来自Spark SQL的DataFrame是为了能够支持上述的各种数据类型。 DataFrame支持很多基本的和结构化的类型，查看Spark SQL datatype reference可以查阅到支持的类型列表。除了Spark SQL指南中列出的类型，DataFrame还支持ML Vector类型。 DataFrame可以隐式的或者明确指定的从一个常规的RDD创建。查看后续的示例代码和Spark SQL programming guide可以获得示例。 DataFrame中的列都是命名列。后续的代码中，使用“text”，“features”和“label”等作为列名。

Pipline组件

Transformers

Transformer是包含特征转换器和学习模型的抽象。在技术上，Transformer实现了一个叫做transform()的方法，这个方法一般通过附加一个或多个列来将一个DataFrame转换为另一个DataFrame。例如：

一个特征转换器可以获取一个DataFrame，读取它的列（比如text），通过map经其转换为一个新列（比如特征向量），然后输出一个附加了此列的新的DataFrame。
一个学习模型可以获取一个DataFrame，读取包含特征向量的列，预测每个特征向量的标签，然后输出一个附加了预测标签列的新的DataFrame。

Estimators

Estimator抽象了学习算法或者任何在数据上应用（fit）或训练的算法的概念。在技术上，Estimator实现了一个叫做fit()的方法，这个方法接受DataFrame并产生一个模型，这个模型是一个Transformer。例如，像LogisticRegression这样的学习算法就是一个Estimator，它调用fit()类训练一个LogisticRegressionModel，LogisticRegressionModel是一个模型，因此也是一个Transformer。

pipline组件的属性

Transformer.transform()和Estimator.fit()都是无状态的。在以后，可以通过备选概念支持有状态的算法。
每一个Transformer或者Estimator实例都有唯一的ID，用以指定参数。（下面将要讨论）

Pipline

在机器学习中，运行一系列的算法来从数据进行处理和学习是很常见的。例如，一个简单的文本文档处理工作流可能包含以下阶段：

将每个文档的文本切分成词。
将每个文档的词转换成数值型的特征向量。
使用特征向量和标签学习一个预测模型。

MLlib用Pipline代表了这样的一个工作流，Pipline由一个序列的PipelineStages（Transformers和Estimators）组成，并按序列顺序来运行。我们将在本节中使用这个简单的工作流作为一个运行示例。

Pipline如何工作

Pipline是由一序列的阶段（stage）组成，每一个阶段是一个Transformer或者Estimator。这些阶段都按照顺序运行，输入的DataFrame在每一个阶段中传递并进行相应的转换。对于Transformer阶段，将在DataFrame上调用transform()方法。对于Estimator阶段，将调用fit()方法并产生一个Transformer（这个Transformer将成为PiplineModel的一部分，或者应用到Pipline），并且在DataFrame上调用这个Transformer的transform()方法。我们举一个简单的文本文档工作流进行说明。下图是一个作为训练过程的Pipline的用法。

在上图中，上面一行代表Pipline的三个stage阶段。前面两个（Tokenizer和HashingTF）是Transformer（蓝色），第三个（LogisticRegression）是一个Estimator（红色）。下面一行代表pipline的数据流，圆柱代表DataFrame。Pipline.fit()方法在原始DataFrame上被调用，这个原始DataFrame有一行行的文本文档和标签。Tokenizer.transform()方法将一行行的文本文档切分成词，向DataFrame添加了一个以词为列的新的列。HashingTF.transform()方法将词列转换为特征向量，向DataFrame添加了一个以这些向量为列的新列。现在，因为LogisticRegression是一个Estimator，Pipline首先调用LogisticRegression.fit()来产生一个LogisticRegressionModel。如果这个Pipline有更多的阶段，它将在把这个DataFrame传给下一个阶段之前调用LogisticRegressionModel的transform()方法。在Pipline和PiplineModel的帮助下，训练和测试数据能够经历相同的特征处理步骤。

细节

DAG Pipline：Pipline的stage阶段被指定为有序的数组。这里给出的示例都是线性的Pipline，例如，在这种Pipline中，每个阶段使用的数据都是从前一个阶段产生。当数据流图形成了有向无环图，就可能产生非线性Pipline。这种图目前是隐式的指定每个阶段输入和输出列的名字（通常指定为参数）。如果Pipline形成DAG，那么各个阶段必须被指定为拓扑有序的。运行时检查：自从Pipline可以在DataFrame上操作不同的类型后，Pipline就不能使用编译时类型检查。在实际运行Pipline之前，Pipeline和PipelineModel将执行运行时检查。这种类型检查是使用DataFrame的schema来完成的，schema是DataFrame中列的数据类型的一种描述。唯一Pipline阶段：Pipline的阶段都应该是唯一的实例。例如，相同的实例myHashingTF不能连续两次插入到Pipline中因为Pipline的stage阶段都要有唯一的ID。然而不同的实例myHashingTF1和myHashingTF2（都是HashingTF）可以被插入到同一个Pipline，因为不同的实例在创建时会有不同的ID。

参数

MLlib的Estimator和Transformer使用同样的API来指定参数。 Param是一个带有自包含文档的命名参数。ParamMap是（参数，值）对的一个集合。这里有两种常用途径将参数传递给一个算法：

为一个实例设置参数。例如，如果lr是LogisticRegression的一个实例，可以调用lr.setMaxIter(10)来使得lr.fit()最多进行10次迭代。这种API类似spark.mllib包里使用的API。
将ParamMap传给fit()或transform()。ParamMap中任何参数都会覆盖之前通过setter方法指定的参数值。

参数属于特定的Estimator和Transformer。例如，如果我们有两个LogisticRegression实例lr1和lr2，我们可以建立一个ParamMap来指定各自的maxIterable参数：ParamMap(lr1.maxIter -> 10, lr2.maxIter -> 20)。在两个都有maxIter参数的算法在同一个Pipline中时，这种方式很有用。