机器学习及flinkML算法

机器学习概念

机器学习算法根据训练数据(training data)使得表示算法行为的数学目标最大化，并以此来进行预测或者做出决定。机器学习分为分类、回归、聚类等，每种都有不一样的目标。

应用场景和处理流程

• 所有的算法都需要定义每个数据点的特征(feature)集->输入；
• 正确的定义特征才是机器学习中最有挑战的部分。
• 大多数算法都是专为数据特征(就是一个代表各个特征值的数字向量)定义的，因此提取特征并转化为特征向量是机器学习过程中重要的一步。
• 输入数据分为“训练集”和“测试集”，并且只使用前者进行训练，这样就可以用后者来检验模型是否过度拟合了训练数据。
• 机器学习流水线会训练出多个不同版本的模型，然后分别对其进行评估。Ml提供几个算法进行模型评估。

分类算法

基于已经被标注的其他数据点作为例子来识别一个数据点属于几个类别中的哪一种；比如判断一封邮件是否为垃圾邮件。
垃圾邮件分类做法:

1. HashingTF

文本数据构建词频特征向量

2. LogisticRegressionWithSGD

使用随机梯度下降法实现逻辑回归。

---

监督学习

• SVM使用通信高效的分布式双坐标上升（CoCoA）
• 多元线性回归
• 优化框架
• L-BFGS
• Generalized Linear Models
• Multiple linear regression
• LASSO, Ridge regression
• Multi-class Logistic regression
• Random forests
• Support Vector Machines
• Decision trees

---

无监督学习

• k-最近邻居关联
• Unsupervised learning
• Clustering
• K-means clustering
• Principal Components Analysis
• Recommendation
• ALS
• Text analytics
• LDA

---

数据预处理

• 多项式特征
• 标准尺度
  MinMax Scaler

---

降维处理

• 模型选择和性能评估
• 多种评分功能模型评估
• 模型选择和评估的交叉验证

---

flinkML目前支持的算法

监督学习

• 基于通信效率的SVM分布式双坐标提升(CoCoA)
• 多元线性回归
• 优化框架

非监督学习

• kNN(K最邻近算法)算法

数据预处理

• 多项式特征
• 标准定标器
• 极大极小标量

推荐算法

• 交替最小二乘法(ALS)

离群值选择

• 随机离群点选择

公用事业公司

• 距离度量
• 交叉验证

分类代码开始

• 添加FlinkML依赖到 pom.xml.

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-ml_2.11</artifactId>
  <version>1.8.0</version>
</dependency>

1. 导入数据

• 格式化数据 比如 LibSVN ，监督学习问题一般使用LabeledVector类来表示(Label,Features)
• 下载UCI ML仓库数据集，该数据集“包含了一项关于乳腺癌手术患者存活率的研究的案例”，数据格式如下所示：

30,64,1,1
30,62,3,1
30,65,0,2
31,59,2,1

前三列代表特征，最后一列代表分类，the 4th column indicates whether the patient survived 5 years or longer (label 1), or died within 5 years (label 2)

import org.apache.flink.api.scala._

val env = ExecutionEnvironment.getExecutionEnvironment

val survival = env.readCsvFile[(String, String, String, String)]("/path/to/haberman.data")

将数据转换为DataSet[LabeledVector],用FlinkML分类算法，第四列是分类标签，构建LabeledVector元素如下

import org.apache.flink.ml.common.LabeledVector
import org.apache.flink.ml.math.DenseVector

val survivalLV = survival
  .map{tuple =>
    val list = tuple.productIterator.toList
    val numList = list.map(_.asInstanceOf[String].toDouble)
    LabeledVector(numList(3), DenseVector(numList.take(3).toArray))
  }

然后我们可以用这些数据来训练学习者。不过，我们将使用另一个数据集来举例说明如何构建学习者;如何导入其他数据集格式，举例如下：
LibSVM 文件
ML数据集的一种常见格式是LibSVM格式。FlinkML通过MLUtils对象提供的readLibSVM函数，提供了使用LibSVM格式加载数据集的实用程序。您还可以使用writeLibSVM函数以LibSVM格式保存数据集。
让我们导入svmguide1数据集。你可以在这里下载训练集和测试集。这是一个天文粒子二分类数据集，由Hsu等人在他们的实用支持向量机(SVM)指南中使用。它包含4个数值特征，以及类标签。
导入数据集合如下：

import org.apache.flink.ml.MLUtils

val astroTrainLibSVM: DataSet[LabeledVector] = MLUtils.readLibSVM(env, "/path/to/svmguide1")
val astroTestLibSVM: DataSet[LabeledVector] = MLUtils.readLibSVM(env, "/path/to/svmguide1.t")

2. 数据分类
   导入训练和测试数据集后，为分类做好准备。由于Flink SVM只支持阈值为+1.0和-1.0的二进制值，加载LibSVM数据集后需要进行转换，因为它使用了1和0进行标记。
   转换可以用一个简单的归一化映射函数:

import org.apache.flink.ml.math.Vector

def normalizer : LabeledVector => LabeledVector = { 
    lv => LabeledVector(if (lv.label > 0.0) 1.0 else -1.0, lv.vector)
}
val astroTrain: DataSet[LabeledVector] = astroTrainLibSVM.map(normalizer)
val astroTest: DataSet[(Vector, Double)] = astroTestLibSVM.map(normalizer).map(x => (x.vector, x.label))

转换了数据集后，开始训练一个预测器，如线性SVM分类器。我们可以为分类器设置一些参数。在这里，我们设置Blocks参数，它被使用的底层CoCoA算法用于分割输入。正则化参数决定了l₂正则化的应用数量，用于避免过拟合。步骤大小决定权重向量更新到下一个权重向量值的贡献。此参数设置初始步长。

import org.apache.flink.ml.classification.SVM

val svm = SVM()
  .setBlocks(env.getParallelism)
  .setIterations(100)
  .setRegularization(0.001)
  .setStepsize(0.1)
  .setSeed(42)

svm.fit(astroTrain)

现在我们可以对测试集进行预测，并使用evaluate函数创建(真值、预测)对。

val evaluationPairs: DataSet[(Double, Double)] = svm.evaluate(astroTest)

接下来，我们将了解如何预处理数据，并使用FlinkML的ML管道功能。
3. 数据预处理和管道(pipelines)
当使用SVM分类时，通常增益的预处理步骤是将输入特征缩放到[0,1]范围，以避免极值特征占主导地位。FlinkML有许多转换器(如用于预处理数据的MinMaxScaler)，其中一个关键特性是能够将转换器和预测器链接在一起。这允许我们运行相同的转换管道，并以一种直接且类型安全的方式对火车和测试数据进行预测。您可以在pipeline文档中阅读更多关于FlinkML管道系统的信息。
首先为数据集中的特征创建一个归一化转换器，并将其链接到一个新的SVM分类器。

import org.apache.flink.ml.preprocessing.MinMaxScaler

val scaler = MinMaxScaler()

val scaledSVM = scaler.chainPredictor(svm)

现在，我们可以使用新创建的管道对测试集进行预测。
首先，我们再次调用fit，来训练定标器和SVM分类器。然后将测试集的数据自动缩放，然后将其传递给SVM进行预测。

scaledSVM.fit(astroTrain)

val evaluationPairsScaled: DataSet[(Double, Double)] = scaledSVM.evaluate(astroTest)

按比例的输入应该能给我们更好的预测性能。

FlinkML文档

---

FlinkML算法二-交替最小二乘法（Alternating Least Squares）

交替最小二乘法（ALS）算法将一个给定的R矩阵因式分解为 U 和 V 两个因子，例如 R≈U^TV。 未知的行的维度被用作算法的参数，叫做潜在因子。 由于矩阵因式分解可以用在推荐系统的场景，U和V矩阵可以分别称为用户和商品矩阵。 用户矩阵的第i列用u_i表示，商品矩阵的第i列用v_i表示。 R 矩阵称为评价矩阵可以用(R)_i,j=r_i,j 表示。 为了找到用户和商品矩阵，如下问题得到了解决：
$argmin$
argminU,V∑{i,j∣ri,j≠0}(ri,j−uTivj)2+λ(∑inui∥ui∥2+∑jnvj∥vj∥2)
λ作为正则化因子，n_ui作为用户i评过分的商品数量， n_vj作为商品j被评分的次数。 这个因式分解方案避免了称作加权λ​因式分解的过拟合。

通过修复U 和 V矩阵，我们获得可以直接解析的二次形式。 问题的解决办法是保证总消耗函数的单调递减。通过对U 或 V矩阵的这一步操作，我们逐步的改进了矩阵的因式分解。

R 矩阵作为 (i,j,r) 元组的疏松表示。i 为行索引，j 为列索引，r 为 (i,j) 位置上的矩阵值。

操作

ALS 是一个预测模型（Predictor）。 因此，它支持拟合（fit）与预测（predict）两种操作。

拟合

ALS用于评价矩阵的疏松表示过程的训练：
fit: DataSet[(Int, Int, Double)] => Unit

预测

ALS会对每个元组行列的所有索引进行评分预测：
predict: DataSet[(Int, Int)] => DataSet[(Int, Int, Double)]

参数

ALS的实现可以通过下面的参数进行控制：

参数	描述
NumFactors	底层模型中使用的潜在因子数目。等价于计算用户和商品向量的维度。 (默认值：10)
Lambda	因式分解的因子。 该值用于避免过拟合或者由于强生成导致的低性能。 (默认值：1)
Iterations	最大迭代次数。 (默认值：10)
Blocks	设定用户和商品矩阵被分组后的块数量。块越少，发送的冗余数据越少。然而，块越大意味着堆中需要存储的更新消息越大。如果由于OOM导致算法失败，试着降低块的数量。 (默认值：None)
Seed	用于算法生成初始矩阵的随机种子。 (默认值：0)
TemporaryPath	导致结果被立即存储到临时目录的路径。如果该值被设定，算法会被分为两个预处理阶段，ALS迭代和计算最后ALS半阶段的处理中阶段。预处理阶段计算给定评分矩阵的OutBlockInformation和InBlockInformation。每步的结果存储在特定的目录。通过将算法分为更多小的步骤，Flink不会在多个算子中分割可用的内存。这让系统可以处理更大的独立消息并提升总性能。 (默认值：None)

例子

// 从CSV文件读取输入数据集
val inputDS: DataSet[(Int, Int, Double)] = env.readCsvFile[(Int, Int, Double)](
  pathToTrainingFile)

// 设定ALS学习器
val als = ALS()
.setIterations(10)
.setNumFactors(10)
.setBlocks(100)
.setTemporaryPath("hdfs://tempPath")

// 通过一个map参数设置其他参数
val parameters = ParameterMap()
.add(ALS.Lambda, 0.9)
.add(ALS.Seed, 42L)

// 计算因式分解
als.fit(inputDS, parameters)

// 从CSV文件读取测试数据
val testingDS: DataSet[(Int, Int)] = env.readCsvFile[(Int, Int)](pathToData)

// 通过矩阵因式分解计算评分
val predictedRatings = als.predict(testingDS)