机器学习工作流

一、概念

一个典型的机器学习过程从数据收集开始，要经历多个步骤，才能得到需要的输出。这非常类似于流水线式工作，即通常会包含源数据ETL（抽取、转化、加载），数据预处理，指标提取，模型训练与交叉验证，新数据预测等步骤。

MLlib标准化了用于机器学习算法的API，从而使将多种算法组合到单个管道或工作流程中变得更加容易。 本节介绍了Pipelines API引入的关键概念，其中PipeLine（管道）概念主要受scikit-learn项目的启发。

在介绍工作流之前，我们先来了解几个重要概念：

DataFrame：使用Spark SQL中的DataFrame作为ML数据集，该数据集可以保存各种数据类型。 例如，DataFrame可以具有不同的列，用于存储文本，特征向量，真实标签和预测。

Transformer：翻译成转换器，是一种算法，可以将一个DataFrame转换为另一个DataFrame。 例如，ML模型是一个Transformer，它将具有特征的DataFrame转换为具有预测的DataFrame。

Estimator：翻译成评估器，它是学习算法或在训练数据上的训练方法的概念抽象。在 Pipeline 里通常是被用来操作 DataFrame 数据并生产一个 Transformer。从技术上讲，Estimator实现了一个方法fit（），它接受一个DataFrame并产生一个转换器。例如，诸如LogisticRegression之类的学习算法是Estimator，调用fit（）可以训练LogisticRegressionModel，后者是Model，因此是Transformer。

Parameter：Parameter 被用来设置 Transformer 或者 Estimator 的参数。现在，所有转换器和估计器可共享用于指定参数的公共API。ParamMap是一组（参数，值）对。

PipeLine：翻译为工作流或者管道。管道将多个“变形器”和“估计器”链接在一起，以指定ML工作流程，并获得结果输出。 例如，简单的文本文档处理工作流程可能包括几个阶段：
1、将每个文档的文本拆分为单词。
2、将每个文档的单词转换成数字特征向量。
3、使用特征向量和标签学习预测模型。
MLlib将这样的工作流表示为“管道”，它由要按特定顺序运行的一系列PipelineStages（变压器和估计器）组成。

二、工作原理

要构建一个 Pipeline工作流，首先需要定义 Pipeline 中的各个工作流阶段PipelineStage，（包括转换器和评估器），比如指标提取和转换模型训练等。有了这些处理特定问题的转换器和 评估器，就可以按照具体的处理逻辑有序的组织PipelineStages 并创建一个Pipeline。比如：

Pipeline pipeline = new Pipeline().setStages(new  PipelineStage[]{tokenizer,hashingTF,lr});

然后就可以把训练数据集作为输入参数，调用 Pipeline 实例的 fit 方法来开始以流的方式来处理源训练数据。这个调用会返回一个 PipelineModel 类实例，进而被用来预测测试数据的标签。更具体的说，工作流的各个阶段按顺序运行，输入的DataFrame在它通过每个阶段时被转换。 对于Transformer阶段，在DataFrame上调用transform（）方法。 对于估计器阶段，调用fit（）方法来生成一个转换器（它成为PipelineModel的一部分或拟合的Pipeline），并且在DataFrame上调用该转换器的transform（）方法。

Pipeline1

上面，顶行表示具有三个阶段的流水线。 前两个（Tokenizer和HashingTF）是Transformers（蓝色），第三个（LogisticRegression）是Estimator（红色）。 底行表示流经管线的数据，其中圆柱表示DataFrames。 在原始DataFrame上调用Pipeline.fit（）方法，它具有原始文本文档和标签。 Tokenizer.transform（）方法将原始文本文档拆分为单词，向DataFrame添加一个带有单词的新列。 HashingTF.transform（）方法将字列转换为特征向量，向这些向量添加一个新列到DataFrame。 现在，由于LogisticRegression是一个Estimator，Pipeline首先调用LogisticRegression.fit（）产生一个LogisticRegressionModel。 如果流水线有更多的阶段，则在将DataFrame传递到下一个阶段之前，将在DataFrame上调用LogisticRegressionModel的transform（）方法。

值得注意的是，工作流本身也可以看做是一个估计器。在工作流的fit（）方法运行之后，它产生一个PipelineModel，它是一个Transformer。 这个管道模型将在测试数据的时候使用。 下图说明了这种用法。

Pipeline2

在上图中，PipelineModel具有与原始流水线相同的级数，但是原始流水线中的所有估计器都变为变换器。 当在测试数据集上调用PipelineModel的transform（）方法时，数据按顺序通过拟合的工作流。 每个阶段的transform（）方法更新数据集并将其传递到下一个阶段。工作流和工作流模型有助于确保培训和测试数据通过相同的特征处理步骤。

三、代码实现

以逻辑斯蒂回归为例，构建一个典型的机器学习过程，来具体介绍一下工作流是如何应用的。我们的目的是查找出所有包含”spark”的句子，即将包含”spark”的句子的标签设为1，没有”spark”的句子的标签设为0。

3.1、构建训练数据集

import java.util.Arrays;
import java.util.List;
import org.apache.spark.ml.Pipeline;
import org.apache.spark.ml.PipelineModel;
import org.apache.spark.ml.PipelineStage;
import org.apache.spark.ml.classification.LogisticRegression;
import org.apache.spark.ml.feature.HashingTF;
import org.apache.spark.ml.feature.Tokenizer;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
SparkSession spark = SparkSession.builder().appName("MLPipelines").master("local").getOrCreate();
//构建训练数据集
List<Row> data = Arrays.asList(RowFactory.create(0L, "a b c d e spark", 1.0),
                               RowFactory.create(1L, "b d", 0.0),
                               RowFactory.create(2L, "spark f g h", 1.0),
                               RowFactory.create(3L, "hadoop mapreduce", 0.0));
System.out.println(data);
/**
*控制台输出结果：
-------------------------------------------------------------------------------------
[[0,a b c d e spark,1.0], [1,b d,0.0], [2,spark f g h,1.0], [3,hadoop mapreduce,0.0]]
-------------------------------------------------------------------------------------
**/
StructType schema = new StructType(new StructField[] {
    new StructField("id",DataTypes.LongType,false,Metadata.empty()),
    new StructField("text", DataTypes.StringType, false, Metadata.empty()),
    new StructField("label", DataTypes.DoubleType, false, Metadata.empty()),
});
Dataset<Row> training = spark.createDataFrame(data,schema);
training.show(false);
/**
*控制台输出结果：
    +---+----------------+-----+
    |id |text            |label|
    +---+----------------+-----+
    |0  |a b c d e spark |1.0  |
    |1  |b d             |0.0  |
    |2  |spark f g h     |1.0  |
    |3  |hadoop mapreduce|0.0  |
    +---+----------------+-----+
**/

3.2、定义 Pipeline 中的各个工作流阶段PipelineStage

在这一步中我们要定义 Pipeline 中的各个工作流阶段PipelineStage，包括转换器和评估器，具体的，包含tokenizer, hashingTF和lr三个步骤。

Tokenizer tokenizer = new Tokenizer().setInputCol("text")
                                     .setOutputCol("words");

HashingTF hashingTF = new HashingTF().setNumFeatures(1000)
                                     .setInputCol(tokenizer.getOutputCol())
                                     .setOutputCol("features");

LogisticRegression lr = new LogisticRegression().setMaxIter(10).setRegParam(0.01);

3.3、创建一个Pipeline

有了这些处理特定问题的转换器和评估器，接下来就可以按照具体的处理逻辑有序的组织PipelineStages 并创建一个Pipeline。

Pipeline pipeline = new Pipeline().setStages(new PipelineStage[]{tokenizer,hashingTF,lr});

3.4、创建模型

现在构建的Pipeline本质上是一个Estimator，在它的fit（）方法运行之后，它将产生一个PipelineModel，它是一个Transformer。

PipelineModel model = pipeline.fit(training);

我们可以看到，model的类型是一个PipelineModel，这个管道模型将在测试数据的时候使用。所以接下来，我们先构建测试数据。

List<Row> testRaw = Arrays.asList(RowFactory.create(4L, "spark i j k"),
        RowFactory.create(5L, "l m n"),
        RowFactory.create(6L, "spark a"),
        RowFactory.create(7L, "apache hadoop")
        );
Dataset<Row> test = spark.createDataFrame(testRaw,schema);
test.select("id", "text").show(false);
/**
*控制台输出结果：
    +---+-------------+
    |id |text         |
    +---+-------------+
    |4  |spark i j k  |
    |5  |l m n        |
    |6  |spark a      |
    |7  |apache hadoop|
    +---+-------------+
**/

3.5、预测结果

然后，我们调用我们训练好的PipelineModel的transform（）方法，让测试数据按顺序通过拟合的工作流，生成我们所需要的预测结果。

model.transform(test).select("id",  "text", "probability",  "prediction").show(false);
/**
    *控制台输出结果：
   +---+--------------+----------------------------------------+----------+
   |id |text          |probability                             |prediction|
   +---+--------------+----------------------------------------+----------+
   |4  |spark i j k   |[0.540643354485232,0.45935664551476796] |0.0       |
   |5  |l m n         |[0.9334382627383527,0.06656173726164716]|0.0       |
   |6  |spark a       |[0.1504143004807332,0.8495856995192668] |1.0       |
   |7  |apache  hadoop|[0.9768636139518375,0.02313638604816238]|0.0       |
   +---+--------------+----------------------------------------+----------+
**/

通过上述结果，我们可以看到，第4句和第6句中都包含”spark”，其中第六句的预测是1，与我们希望的相符；而第4句虽然预测的依然是0，但是通过概率我们可以看到，第4句有46%的概率预测是1，而第5句、第7句分别只有7%和2%的概率预测为1，这是由于训练数据集较少，如果有更多的测试数据进行学习，预测的准确率将会有显著提升。

参考资料: http://spark.apache.org/docs/latest/ml-pipeline.html