Flink ML学习笔记

概念

Flink ML是一个机器学习算法库。用户可以使用标准机器学习API或者自定义机器学习算法来进行离线或在线的模型训练和预测服务。

一、环境准备

• 安装1.15及以上版本的flink到本地环境
• 导入flink table api相关依赖

二、开发中的基本概念

1、Table API

Flink ML的API基于Flink Table API。Table API是一个针对Java、Scala和Python的语言集成查询API，它允许以非常直观的方式组合来自关系运算符（如选择、筛选和连接）的查询。Table API允许使用多种数据类型（可参考Flink文档数据类型）。除了这些类型之外，Flink ML还提供对Vector类型的支持。Table API与Flink DataStream API无缝集成，可以轻松地在所有基于它们的API和库之间切换。

2、Stage

Stage是Pipeline或Graph中的节点。它是Flink ML中的基本组件。该接口只是一个概念，没有任何实际功能。其子类包括：

A、Estimator：

Estimator是机器学习算法中负责训练过程的Stage。它实现了一个fit（）方法，用来获取一个表列表并生成一个Model。

B、AlgoOperator：

AlgoOperators用于编码通用的多输入多输出计算逻辑。它实现了一个transform（）方法，对给定的输入表应用特定的计算逻辑，并返回结果表列表。

C、Transformer：

Transformer是一个AlgoOperator，其语义差异在于它编码转换逻辑，因此输出中的一个记录通常对应于输入中的一条记录。相比之下，AlgoOperator更适合表达聚合逻辑，其中输出中的记录可以从输入中的任意数量的记录中计算出来。

D、Model：

Model是一个带有额外API的Transformer，用于设置和获取模型数据。它通常是通过在表列表上拟合Estimator生成的。它提供了getModel数据（）和setModelData（），允许用户显式地将模型数据表读写到Transformer中。每个表可以是一个无限的模型数据更改流。

Stage的一个典型用法是首先创建一个Estimator实例，通过调用它的fit（）方法触发它的训练过程，然后使用生成的Model实例执行预测。

3、Builders

为了将Flink ML阶段组织成更复杂的格式，以实现高级功能，例如将数据处理和机器学习算法链接在一起，Flink ML提供了API，帮助管理Flink作业中各Stage的关系和结构。这些API的入口包括Pipeline和Graph。

A、Pipeline：

Pipeline充当Estimator。它由一个有序的Stage列表组成，每个Stage可以是一个Estimator、Model、Transformer或AlgoOperator。它的fit（）方法按顺序遍历该Pipeline的所有Stage，并在每个Stage执行以下操作，直到最后一个Estimator（含）。
如果Stage是一个Estimator，它将使用输入表调用阶段的fit（）方法来生成模型。如果在这个Stage之后有Estimator，它将使用生成的模型转换输入表以获得结果表，然后将结果表作为输入传递到下一Stage。
如果一个Stage是AlgoOperator，并且在这个Stage之后有Estimator，它将使用这个Stage转换输入表以获得结果表，然后将结果表作为输入传递到下一个Stage。

在所有的Estimator都经过训练以适合其输入表之后，将使用该Pipeline中的相同Stage创建一个新的PipelineModel，除了PipelineModel中的所有Estimator被上述过程中生成的Model替换。
PipelineModel充当Model。它由一个有序的Stage列表组成，每个Stage可以是Model、Transformer或AlgoOperator。它的transform（）方法将此PipelineModel中的所有Stage按顺序应用于输入表。一级的输出用作下一级的输入（如果有）。最后一个Stage的输出作为此方法的结果返回。
可以通过向Pipeline的构造函数传递Stage列表来创建Pipeline。

三、迭代算法

迭代算法是ML库的基本构建块。在机器学习算法中，迭代算法可以用于离线或在线训练过程。通常，需要两种类型的迭代算法，Flink ML支持这两种迭代，以便为各种算法提供基础结构。

1、有界迭代：

通常用于离线情况，在这种情况下，算法通常在有界数据集上训练，它更新多轮的参数，直到有界数据集结束。

2、无界迭代：

通常用于在线情况，在这种情况下，算法通常在无界数据集上训练，它累积一小批数据，然后对参数进行一次更新。

迭代算法具有以下行为模式：

迭代算法具有迭代体，该迭代体被重复调用，直到达到某些终止标准（例如，在达到用户指定的时期数之后）。迭代体是实现例如迭代机器学习算法的计算逻辑的算子的子图，其输出可以作为该子图的输入反馈。
在每次调用中，迭代主体基于用户提供的数据以及最新的模型参数更新模型参数。
迭代算法将用户提供的数据和初始模型参数作为输入。
迭代算法可以输出任意用户定义的信息，例如每个历元后的损失或最终模型参数。

因此，迭代算法的行为可以用以下迭代范式来作为特征：

迭代体是具有以下输入和输出的Flink子图：
输入：模型变量（作为数据流的列表）和用户提供的数据（作为另一个数据流列表）
输出：反馈模型变量（作为数据流列表）和用户观察的输出（作为数据流量列表）
一个终止条件，指定迭代体的迭代执行何时终止。
为了执行迭代体，用户需要使用以下输入执行迭代体并获得以下输出。
输入：初始模型变量（作为有界数据流的列表）和用户提供的数据（作为数据流列表）
输出：迭代体发出的用户观察输出。
需要注意的是，迭代体预期的模型变量与用户提供的初始模型变量不同。相反，模型变量被计算为反馈模型变量（由迭代体发出）和初始模型变量（迭代体的调用者提供）的并集。Flink ML提供了实用程序类（参见迭代算法），以使用用户提供的输入运行迭代体。
下图总结了上述迭代范例：

Flink ML迭代的主要条目位于Iterations类中。它主要提供两种公共方法，用户可以根据输入数据是有界还是无界来选择使用其中一种方法。
初始变量流和初始数据流中的所有记录都具有epoch=0。
对于由该运算符发射到非反馈流中的任何记录，该发射记录的历元=触发该发射的输入记录的历。如果此记录由onEpochWatermarkIncremented（）发出，则此记录的epoch=epochWatermark。
对于由该运算符发射到反馈变量流中的任何记录，发射记录的历元=触发该发射的输入记录的历元+1。
该框架将在每个epoch结束时向实现IterationListener的运算符和UDF发送通知。

四、数据类型

Flink ML支持Flink Table API支持的所有数据类型，以及Vector数据类型。
Flink ML支持双值向量。Flink ML中的向量可以是密集的（DenseVector）或稀疏的（SparseVector），这取决于用户如何根据向量的稀疏度创建向量。每个向量以固定大小初始化，用户可以获取或设置向量中任何基于0的索引位置的双倍值。
Flink ML还有一个名为Vectors的类，提供用于实例化向量的实用方法。

五、算子

（一）、分类（classification）

1、KNN：

K近邻（KNN）是一种分类算法。KNN的基本假设是，如果所提供样本的大多数最近K个邻居属于同一标签，那么所提供样本也很可能属于该标签。

2、Linear SVM（Linear Support Vector Machine）：

线性支持向量机（Linear Support Vector Machine，线性SVC）是一种试图找到超平面以最大化分类样本之间距离的算法。

3、Logistic Regression：

Logistic回归是广义线性模型的一个特例。它被广泛用于预测二元响应。

4、Naive Bayes：

朴素贝叶斯是一种多类分类器。基于贝叶斯定理，它假设每对特征之间都有很强的（天真的）独立性。

（二）、聚类（clustering）

1、Kmeans：

K-means是一种常用的聚类算法。它将给定的数据点分组为预定义数量的集群。

（三）、评估器（Evaluator）

1、Binary Classification Evaluator：

二进制分类评估器计算二进制分类的评估度量。输入数据具有rawPrediction、label和可选的权重列。rawPrediction可以是double类型（二进制0/1预测，或label 1的概率）或vector类型（原始预测、得分或label概率的长度-2向量）。输出可能包含由参数MetricsNames定义的不同度量。

（四）、功能工程（Feature Engineering）

1、Bucketizer：

Bucketizer是一种将多列连续特征映射到多列离散特征（即桶索引）的算法。索引在[0，numSplitsInThisColumn-1]中。

2、Min Max Scaler：

最小-最大缩放器是一种将特征值重新缩放到用户定义的公共范围[Min，Max]的算法。

3、One Hot Encoder：

一种热编码将表示为标签索引的分类特征映射到二值向量，该二值向量最多具有单个一个值，该值指示所有特征值集合中特定特征值的存在。这种编码允许期望连续特征（如Logistic回归）的算法使用分类特征。
OneHotEncoder可以转换多个列，为每个输入列返回一个热编码输出向量列。

4、Standard Scaler：

标准缩放器是一种通过去除平均值并将每个维度缩放为单位方差来标准化输入特征的算法。

5、String Indexer：

字符串索引器将输入的一列或多列（字符串/数值）映射到一列或多列索引输出列（整数值）。如果两个数据点对应的输入列相同，则其输出索引相同。索引位于[0，numDistinctValuesInThisColumn]中。
IndexToStringModel使用StringIndexer计算的模型数据将输入索引列转换为字符串列。这是StringIndexerModel的反向操作。

6、Vector Assembler：

Vector Assembler将给定的输入列列表组合成一个向量列。输入列的类型必须是矢量或数值。

（五）、回归（Regression）

1、Linear Regression：

线性回归是一种通过对标量响应和一个或多个解释变量之间的关系进行建模的回归分析。