第2.4节 Pipeline与PipelineModel

Alink提供了Pipeline/PipelineModel，其在功能和使用方式上与Scikit-learn和SparkML的Pipeline/PipelineModel类似。保持训练和预测过程中数据处理的一致性，这样调用过程清晰、简练，也降低了用户的学习成本。Alink Pipeline/PipelineModel与批式/流式组件（BatchOperator/ StreamOperator）一样，将Flink Table作为计算输入和输出结果的数据类型。

Scikit-learn和SparkML的Pipeline是针对批式数据的训练和预测设计的。Alink不仅支持批式数据的训练和预测，也支持将批式训练出的模型用于预测流式数据。

2.4.1 概念和定义

管道（Pipeline）的概念源于Scikit-learn。可以将数据处理的过程看成数据在“管道”中流动。管道分为若干个阶段（PipelineStage），数据每通过一个阶段就发生一次变换，数据通过整个管道，也就依次经历了所有变换。

如果要在管道中加入分类器，对数据进行类别预测，就涉及模型的训练和预测，这需要分两个步骤完成。所以，管道也会被细分为管道定义与管道模型（PipelineModel）。在管道定义中，每个PipelineStage会按其是否需要进行模型训练，分为估计器（Estimator）和转换器（Transformer）。随后，可以对其涉及模型的部分，即对估计器（Estimator）进行估计，从而得到含有模型的转换器。该转换器被称为Model，并用Model替换相应的估计器，从而每个阶段都可以直接对数据进行处理。我们将该Model称为PipelineModel。

上面介绍了几个概念，用图形表示它们之间的关系如下：

• 转换器（Transformer）：Pipeline中的处理模块，用于处理Table。输入的是待处理的数据，输出的是处理结果数据。比如，向量归一化是一个常用的数据预处理操作。它就是一个转换器，输入向量数据，输出的数据仍为向量，但是其范数为1。
• Model：派生于转换器（Transformer）。其可以存放计算出来的模型，用来进行模型预测。其输入的是预测所需的特征数据，输出的是预测结果数据。
• 估计器（Estimator）：估计器是对输入数据进行拟合或训练的模型计算模块，输出适合当前数据的转换器，即模型。输入的是训练数据，输出的是Model。

Pipeline与PipelineModel构成了完整的机器学习处理过程，可以分为三个子过程：定义过程，模型训练过程，数据处理过程。

• 定义过程：按顺序罗列Pipeline所需的各个阶段。Pipeline由若干个Transformer和Estimator构成，按用户指定的顺序排列，并在逻辑上依次执行。
• 模型训练过程：使用Pipeline的fit方法，对Pipeline中的Estimator进行训练，得到相应的Model。Pipeline执行fit方法后得到的结果是PipelineModel。
• 数据处理过程：该过程指的是通过PipelineModel直接处理数据。

比如，LR分类算法作为一个Estimator，可以在构建Pipleline的时候进行定义，之后在fit的过程中，会使用LR的训练算法，得到LR model，并将其作为整个PipelineModel的一部分；在使用PipelineModel处理数据时，会相应地调用LR算法的预测部分。

2.4.2 深入介绍

以使用朴素贝叶斯算法进行多分类为例，演示三个经典场景：批式训练、批式预测与流式预测。为了更好地体现它们之间的关联，我们用一张图表示了出来，如下图所示。

此图有如下特点：

（1）左边这列组件展示了批式训练的流程，中间那列组件展示了批式预测的流程，右边那列组件展示了流式预测的流程。

（2）训练和预测都经历了四个阶段：缺失值填充、MultiStringIndexer（将字符串数据用索引值替换）、VectorAssembler（将各数据字段组装为向量）、使用朴素贝叶斯算法进行训练/预测。

（3）在VectorAssembler阶段，运行时不需要额外的信息。在批式训练、批式预测和流式预测的过程中，组件均可直接使用。

（4）除VectorAssembler外的三个阶段，都需要对整体数据进行扫描或者迭代训练，得到模型后，才能处理（模型预测）数据。训练过程需要拿到所有的模型，缺失值填充和MultiStringIndexer阶段需要在得到模型后，对当前数据进行预测，将预测结果数据传给后面的阶段。

上面的四个阶段分别对应四个PipelineStage，这些PipelineStage构成了Pipeline。对应的Java代码如下：

Pipeline pipeline = new Pipeline()
   .add(
      new Imputer()
      .setSelectedCols("...")
      .setStrategy(Strategy.VALUE)
      .setFillValue("null")
   )
   .add(
      new MultiStringIndexer()
      .setSelectedCols("...")
   )
   .add(
      new VectorAssembler()
      .setSelectedCols("...")
      .setOutputCol("vec")
   )
   .add(
      new NaiveBayesTextClassifier()
      .setVectorCol("vec")
      .setLabelCol("label")
      .setPredictionCol("pred")
   );

Pipeline可多次复用，不同的训练数据通过同样的流程可得到不同的模型：

CsvSourceBatchOp trainData1 = new CsvSourceBatchOp();
CsvSourceBatchOp trainData2= new CsvSourceBatchOp();

PipelineModel model = pipeline.fit(trainData1);
PipelineModel model2 = pipeline.fit(trainData2);

得到的PipelineModel可重复应用于不同的预测数据：

CsvSourceBatchOp predictData1= new CsvSourceBatchOp();
CsvSourceBatchOp predictData2= new CsvSourceBatchOp();

BatchOperator<?> predict1 = model.transform(predictData1);
BatchOperator<?> predict2 = model.transform(predictData2);

PipelineModel可以无区别地应用于批式预测和流式预测：

CsvSourceBatchOp batchData = new CsvSourceBatchOp();
CsvSourceStreamOp streamData= new CsvSourceStreamOp();

BatchOperator<?> predictBatch = model.transform(batchData);
StreamOperator<?> predictStream = model.transform(streamData);

从PipelineModel可以得到用于本地预测的LocalPredictor，直接嵌入Java应用服务：

String inputSchemaStr = "...";
LocalPredictor localPredictor = model.collectLocalPredictor(inputSchemaStr);

Row inputRow = Row.of("...");
Row pred = localPredictor.map(inputRow);