高级分析是指各种旨在发现数据规律，或根据数据做出预测和推荐等核心问题的技术。机器学习最佳的模型结构要根据要执行的任务制定，最常见的任务包括：

  • 监督学习，包括分类和回归，其目标是根据数据项的各种特征预测每个数据项的标签。
  • 推荐系统，根据行为向用户推荐产品。
  • 无监督学习，包括聚类，异常检测，以及主题建模，其目的是发现数据中的结构。
  • 图分析任务，如发现社交网络中的模式。

1.1 监督学习

监督学习可能是最常见的机器学习任务，它的目标很简单：使用有标签的历史数据（通常被称为因变量）来训练模型，基于该模型和新数据点的各特征来预测该数据点的标签。一个例子是根据一个人的年龄（特征）预测他的收入（因变量）。它的训练过程一般是通过梯度下降这种优化算法实现的，训练算法从一个初始基本模型开始，并且在每次迭代期间会调整模型的各参数来逐渐提升模型准确度。

这一过程的结果是一个训练好的模型，可以利用它来对新数据进行预测。在训练和做出预测的过程中需要完成很多不同的任务，如在使用模型之前需要测试训练好的模型的效果，其基本原理很简单：基于历史数据进行训练，确保它在未训练过的数据上的泛化性，然后在新数据上预测。

1.1.1 分类

分类是监督学习的一种常见任务，它是训练一个算法来预测因变量的类别（属于离散的，有限的一组值）。最常见的情况是二元分类，模型预测一个给定的项属于两组中的哪一个。典型的例子是辨别垃圾邮件。使用一组已经确定为垃圾邮件或非垃圾邮件的历史电子邮件，训练一个模型来分析这些邮件其中的单词和各种属性，并对它们进行预测。一旦模型的性能让人满意，就使用该模型来对模型从未见过的邮件进行预测，自动判别它是否是垃圾邮件。

当分类的项多于两类时，称其为多分类问题。例如有四种不同类型的电子邮件（而不再是之前提到的两个类别）：

  • 垃圾邮件
  • 私人邮件
  • 工作相关邮件
  • 其他

当然还有很多分类任务的应用场景，包括：

  • 预测疾病：医生和医院可能有患者的行为和生理特征的历史数据集，他们可以用这份历史数据集来训练模型（在应用之前需要评估它的准确度和伦理问题），然后利用它来预测患者是否有心脏疾病。这可以是个二分类问题（健康或不健康），也可以是一个多分类问题（健康的心脏，或不同的疾病之一）。
  • 图像分类：有许多来自像苹果、谷歌、Facebook 公司的应用，它们通过用户过往上传的人物图片训练出的一个分类模型，来预测给定一张图片中的人物。另一种常见应用场景是图片的分类或标注图片中的对象。
  • 预测客户流失：预测哪些客户可能不再使用这项服务，可以基于已经流失和未流失的客户数据集，训练出一个二元分类模型，并用它来预测当前客户是否有可能会流失。
  • 公司经常想预测网站的访问者会不会购买某产品。为了实现这种预测，它们可能会用到用户浏览模式或者用户位置等特征。

1.1.2 回归

在分类问题中因变量是一系列离散的值，而在回归问题中要预测连续的变量（实数）。最简单的情况是想预测出某一个实数值而非一个类别，其余的过程大致相同，这就是为什么它们两个都是监督学习，要基于历史数据来预测从未见过的数据。

回归有很多典型的应用场景：

  • 销售预测：商店想利用历史销售数据来预测商品销售情况。销售情况可能会和很多因素有关，也就是说可能会基于很多输入变量，但是一个简单的例子就是基于上周的销售数据来预测下周的数据。
  • 身高预测：根据父母的身高，要预测他们孩子的身高。
  • 预测节目的观众人数：像 Netflix 这样的流媒体公司要预测会有多少用户观看某一个节目。

1.1.3 推荐系统

推荐系统是高级分析最直观的应用之一，通过研究用户对多种商品的显式偏好（通过评级）或隐式偏好（通过观察到的行为），基于用户之间的相似性或商品之间的相似性来推荐给用户他们可能喜欢的商品。通过查看这些相似性，推荐系统可以把相似用户喜欢的商品推荐给他，或者把相似用户已经购买的商品推荐给他。推荐系统是 Spark 任务中很常见的应用场景，并且 Spark 非常适合处理大数据推荐。

下面是一些推荐系统的应用场景：
电影推荐：Netflix 虽然不一定使用 Spark 来处理大数据，但 Netflix 确实要为其用户做大规模推荐，它通过研究用户在 Netflix 上看过和不看什么电影的观影记录来做预测。此外，Netflix 公司有可能考虑到用户之间打分模式的相似性。
产品推荐：亚马逊将产品推荐作为提高销售额的主要手段之一。例如，根据购物车中的物品，亚马逊推荐与曾经加入过购物车的类似的其他物品。同样，每一个产品页面上，亚马逊展示出由其他用户购买的同类产品。

1.2 无监督学习

无监督学习是试图在一组给定的数据中寻找模式或发现隐层结构的方法。这不同于监督学习，因为没有因变量（标签）来做预测。

一些使用无监督学习的应用场景包括：

  • 异常检测：鉴于一些常规事件经常发生，用户可能希望当非常规事件发生时给出预警。例如，安全管理人员可能希望当路上出现一个奇怪物体（损坏车辆，溜冰鞋或骑自行车的人）的时候收到通知。
  • 用户分类：给定一组用户的行为数据，可能要更好地了解某些用户与其他用户共享哪些特性。例如，游戏公司可能基于像在某游戏中花费的时间来聚类用户。该模型可能揭示休闲玩家与铁杆玩家完全不同的行为模式，并根据这种差异性，给每个玩家提供不同的建议或奖励。
  • 主题建模：给定一组文件，可以分析其中所含的词组来看看它们之间是否有某种潜在关系。例如，提供一些关于数据分析的网页，主题建模算法可以基于一个主题中比较常见的词，将这些网页标记成机器学习主题、SQL 主题、流处理主题的页面等等。

1.3 图分析

虽然相比于分类和回归，图分析不是很常用，但它是一个强大的工具。从根本上讲，图分析是在研究给定顶点（对象）和边（表示这些对象之间的关系）的结构。例如，顶点可能代表了人与产品，边可能代表购买行为。通过观察顶点和边的属性，可以更好地理解它们和图形整体结构之间的关联。

由于图表达的是关系，任何可以抽象成关系的数据都可以作为图分析的一个应用场景，这包括：

  • 欺诈预测：Capital One 公司采用 Spark 的图分析功能更好的了解欺诈网络，通过使用历史欺诈信息（如电话号码，地址或名字），他们发现欺诈信贷的请求或交易。举例来说，一个欺诈电话号码的两跳范围内（电话拨出方和接听方构成一个边连接，从某电话号码 A 拨出给 B，B 再拨出给 C，则 B 和 C 都处于该电话号码的两跳范围内）的任何用户帐户可能会被认为是可疑的。
  • 异常检测：通过观察个体之间的网络连接方式，可以标记出异常值，以便进行手动分析。例如，如果在数据中通常每个顶点都有 10 条边，而给定的一个顶点只有一条边相连，这可能是值得研究的奇怪现象。
  • 分类：给定一个网络中顶点的属性信息，就可以根据其他点与该点的连接情况来对其他点进行分类。例如，如果某个人被标记为社会网络中的一个有影响力者，可以将其他具有类似网络结构的人归类为有影响力者。
  • 推荐：谷歌的原始网页推荐算法 PageRank 是以分析网站关系来对网页重要性排名的图算法。例如，有很多链接的网页比没有链接的网页更重要。

1.4 机器学习流程

高级分析机器学习过程的流程步骤如下所示：

image.png

流程：

  1. 搜集与预测任务相关的数据。
  2. 清理和检查数据以更好地理解它。
  3. 执行特征工程以使数据以适合的形式为算法使用（例如，将数据转换为数值向量）。
  4. 使用该数据的一部分作为训练集训练一个或多个模型，生成一些候选模型。
  5. 利用从未被用作训练的数据子集来实际客观地评价结果，从而评估比较模型的效果，这可以更好地了解模型到底怎么样。
  6. 利用上述过程的结果和使用模型进行预测、检测异常、或解决更通用的业务难题。

收集合适的数据后将需要清理和检查它，这通常是所谓探索性数据分析或 EDA 的一部分。EDA 一般是指采用交互式查询和可视化的方法，以便更好地了解数据的分布、数据相关性和其他细节。在这个过程中，可能会发现需要删除一些未标注的数据或者可能错误标注的数据。结构化 API 中的许多 Spark 函数都提供一种简单的方式来清洗和报告数据。

收集和清理数据集之后，需要将它转换成适合于机器学习算法一种形式，这通常是数值特征。特征工程可以造就也可以毁掉一个机器学习应用，所以这是要认真对待的一个步骤。特征工程的过程包括各种任务，诸如正则化数据，增加变量来表示其它变量的相互作用，操纵类别变量，并将它们转换为适当的格式以输入到我们的机器学习模型。在 MLlib 中，所有的变量通常必须作为浮点型向量输入（不管他们实际上代表什么）。

现在拥有了历史信息的数据集（例如，垃圾邮件或不是垃圾邮件），也有一个具体的任务（例如，分类垃圾邮件）。下一步将要训练一个模型来根据输入预测正确的输出。在训练过程中，模型的内部参数将根据模型对输入数据的分类效果发生变化。举例来说，要检测垃圾邮件，算法可能会发现某些词的出现更能用来识别垃圾邮件，因此模型中与这些词相关的权重会更高。最后，训练之后的模型发现，某些单词在分类邮件方面比其他词有更大的影响力（因为它们与垃圾邮件有一些相关性）。

训练过程的输出就是所说的模型，然后模型可被用于理解数据或做未来的预测。为了做出预测，需要给模型输入数据，它会基于对这些输入数据的数学运算来产生输出。以分类为例，给定邮件的属性，它会通过比较训练过的历史垃圾邮件和非垃圾邮件来预测该邮件是否为垃圾邮件。当然，应该将数据集分割成多个部分，并且只使用其中一部分进行训练，这是在机器学习过程中的一个重要步骤，因为当建立一个高级分析模型，需要能泛化到它以前没有见过的数据。

将数据集分成多个部分，能够客观地对其中一部分数据（一组未用来训练模型的数据）对训练后模型的有效性进行测试，目标是看模型是否了解有关此数据的本质，或者它是否只注意到特定于训练集的内容（又称为过拟合），这就是它为什么把被称为测试集。在训练模型的过程中，还可能采取另一个单独的数据子集，并将其视为另一种类型的测试集（称为验证集），以便尝试不同的超参数（影响训练过程的参数），并在没有过拟合到测试集的情况下比较相同模型的不同变化。

配置适当的训练集、验证集和测试集对成功运用机器学习解决问题是非常重要的，如果不正确地分割这些数据集，那么就很容易导致过拟合，即模型对新数据的泛化能力弱。

1.5 Mllib

1.5.1 MLlib 概述

Mllib 是基于 Spark 的一个软件包，它提供各种 API 接口用于收集和清理数据、特征工程和特征选择、训练和微调大型有监督和无监督机器学习模型、并在生产中使用这些模型。

MLlib 实际上由两个利用不同核心数据结构的包组成：

  • org.apache.spark.ml 软件包使用 DataFrame 的接口，还提供了用于构建机器学习流程的高层次接口，它将有助于标准化执行上述步骤。
  • 较低级别的软件包 org.apache.spark.mllib，包括 Spark 低级别的 RDD API 接口。它处于维护模式，也就是说处于更新 bug 状态而不会再有新功能。

基于单机有许多用于执行机器学习任务的工具，虽然有几个很好的工具（如 TensorFlow）可供选择，但这些基于单机的工具要么无法训练海量数据，或者处理时间太长。这意味着单机工具通常是 MLlib 的补充，当涉及特别需要扩展性的问题时，就要利用 Spark 的能力。

利用 Spark 的可扩展能力，有两个关键的应用场景：

  • 希望利用 Spark 进行数据预处理和特征生成，以减少从大量数据中生成训练和测试集所需的时间，然后再利用单机机器学习库对这些给定的数据集进行训练。
  • 当输入数据或模型变得太难或不方便在单机上处理时，可以使用 Spark。

一个重要提示是，虽然利用 Spark 可以使训练和数据预处理工作变得很简单，但仍需要牢记一些复杂性，尤其是在部署训练好的模型时。例如，Spark 并没有提供内置的方式来支持模型中的低延迟预测，因此可能希望将处理好的模型导出到另一个系统或自定义程序来做到这一点，MLlib 通常支持导出模型到其他工具。

1.5.2 MLlib 结构类型

在 MLlib 有几个基本的 “结构” 类型：转换器（**transformer**），估计器（estimator**），评估器（evaluator**），和流水线（pipeline**）**。通过结构化的表示，在定义端到端的机器学习流水线时会考虑这些类型，它们将提供共同的语言来定义哪些属于流水线的哪一部分。

下图说明了使用 Spark 开发机器学习模型时需遵循的总体开发流程：

image.png

转换器（transformer）是将原始数据以某种方式进行转换的函数。它可能会（从另外两个变量）创建一个新的变量，对某一列进行归一化，或仅仅将一个 Integer 类型值变为 Double 类型值输入进模型。转换器的一个例子是将字符串类型变量转换为 MLlib 能够使用的数值型值，它主要用于数据预处理和特征工程阶段，以 DataFrame 作为输入并生成一个新的 DataFrame 作为输出，如下图所示：

image.png

估计器（Estimator）可能用来做两件事其中之一。第一，估计器可以作为数据初始化的转换器，例如为了对数值数据进行归一化，需要基于某列中的当前值来初始化转换器，这需要传递两次数据，第一次传递生成初始化值，第二次实际在数据上应用生成函数。第二，基于数据训练模型的算法也称为估计器。

评估器（Evaluator）允许根据某种效果评价指标（如受试者工作特征曲线 - ROC 曲线）来评价给定模型的表现如何。在使用评估器从测试的模型中选择最佳模型之后，就可以使用该模型进行预测。从较高的层次上，可以一个个地指定转换器、估计器和评估器，但是也可以把步骤指定为 pipeline 中的 stage，这种流水线类似于 scikit-learn 中的流水线概念。

除了构建 pipeline 的结构类型外，还有几种低级别的数据类型可能在 MLlib 中使用，Vector 是最常见的。每当将一组特征数据传递到机器学习模型中时，必须将其组织成 Double 类型的向量，此向量可以是稀疏的（其中大多数元素为零）也可以是稠密的（其中有许多非重复值）。向量是以不同的方式创建的，要创建稠密向量需要指定数组中的所有值，要创建稀疏向量可以指定向量的大小、索引和非零元素的值。稀疏向量是大多数值为零时情况下的最好表达形式，因为这是一种更易压缩的表示形式。

下面是如何手动创建向量的示例：

  1. from pyspark.ml.linalg import Vectors
  2. denseVec = Vectors.dense(1.0, 2.0, 3.0)
  3. size = 3
  4. idx = [1, 2] # locations of non-zero elements in vector
  5. values = [2.0, 3.0]
  6. sparseVec = Vectors.sparse(size, idx, values)

1.5.3 MLlib pipeline

前面已经概述了可能会遇到的核心概念，下面就来创建一个简单的 pipeline 来演示每个组件部分，这里将使用一个规模较小的合成数据集，在进一步讨论之前先读取数据：

  1. #in Python
  2. df = spark.read.json("/data/simple-ml")
  3. df.orderBy("value2").show()

这是数据样本：

  1. +-----+----+------+------------------+
  2. |color| lab|value1| value2|
  3. +-----+----+------+------------------+
  4. |green|good| 1|14。386294994851129|
  5. ......
  6. | red| bad| 16|14。386294994851129|
  7. |green|good| 12|14。386294994851129|
  8. +-----+----+------+------------------+

此数据集包含以下标签：好或坏的分类 lab、代表颜色的分类 color，和两个数值变量 value1 和 value2。虽然数据是合成的，但可以想象成这个数据集代表公司客户的健康状况，其中 color 列代表由客服代表所做的某种类别的健康评级，lab 列代表真正的客户健康状况，另外两个值是一些数值型的行为度量（例如在线时间和花费开销）。假设想要训练一个分类模型，希望根据这些值预测一个二元变量（标签）。

除了 JSON，还有一些特定的数据格式用于监督学习，包括 LIBSVM。这些格式具有真正的值标签和稀疏输入数据，Spark 可以使用其 DataSource API 读取和写入这些格式。下面是一个如何使用 DataSource API 的例子：

  1. spark.read.format("libsvm").load("/data/sample_libsvm_data.txt")

如前所述，转换器帮助以某种方式操纵当前的列数据，操纵这些列通常是为了构建特征（会将它们输入到模型中）。转换器可以用来减少特征的数量、添加更多特征、操作当前的列、或简单地帮助纠正数据格式，转换器会向 DataFrame 添加新列。当使用 Mllib 时，所有 Spark 机器学习算法的输入是由 Double 类型（表示标签）和 Vector[Double] 类型（表示特征）组成。当前的例子数据不符合这一要求，因此需要将其转换为正确的格式。

要在上面例子做到这一点，要指定一个 Rformula，它是指定机器学习转换的声明式语言，用来根据数据集自动生成特征（feature）和标签（label），RFormula 支持 R 语言运算符的一个有限子集，它对于实际应用中的简单模型和操作有很好的支持。

基本 RFormula 操作符是：

  • ~：目标（标签）和项（特征）的分隔符号。
  • +：合并项，“+ 0” 表示删除空格。
  • -：删除项，“-1”表示删除空格，和 “+ 0” 起相同作用。
  • :：交互（数值乘法，或类别二值化）。
  • .：除了目标列的全部列。

为了应用具有此语法的转换，需要导入相关的类，然后需要定义公式。在该例子中，希望使用所有列（. 符号），在 value1 和 color 列之间添加交互，在 value2 和 color 列之间之间添加交互，将这些特征视为新特性：

  1. from pyspark.ml.feature import RFormula
  2. supervised = RFormula(formula="lab ~ . + color: value1 + color: value2")

此时以声明方式指定了如何将数据更改为模型训练要使用的格式。接下来的步骤是，将 RFormula 转换器应用到数据上，让它发现每个列的可能值。不是所有的转换器都有这个要求，但因为 RFormula 会自动处理分类变量，它需要确定哪些列是用于分类的、哪些不是、还有分类的列的具体值，出于这个原因要调用 fit 方法。之后，它返回 “训练好” 的转换器，可以使用它来转换数据。现在说明了这些细节所以继续准备 DataFrame：

  1. # in Python
  2. fittedRF = supervised.fit(df)
  3. preparedDF = fittedRF.transform(df)
  4. preparedDF.show()

这是训练和转换过程的输出：

  1. +-----+----+------+------------------+--------------------+-----+
  2. |color| lab|value1| value2| features|label |
  3. +-----+----+------+------------------+--------------------+-----+
  4. |green|good| 1|14。386294994851129|(10，[1，2，3，5，8]，[…1。0|
  5. …| red| bad| 2|14。386294994851129|(10，[0，2，3，4，7]，[…0。0|
  6. +-----+----+------+------------------+--------------------+-----+

在输出中可以看到转换结果，就是名为 features 的列，它包含以前的原始数据。后面发生的事情其实很简单，RFormula 在调用 fit 函数时检查数据，并输出一个根据指定公式转换数据的对象（称为 RformulaModel），这个 “训练好” 的转换器在类型签名中始终称为 Model。当使用这个转换器时，Spark 会自动将分类变量转换为 Double 类型，这样就可以将它输入到一个（尚未指定的）机器学习模型中，特别是它为每个可能的 color 类别分配一个数值，为 color 和 value1/value2 之间的交互变量创建特征，并将它们全部放到一个向量中。然后调用该对象上的 transform 函数，以便将输入数据转换为想要的输出数据。

至此预处理了数据，并添加了一些特征，现在是时候在这个数据集上实际训练模型了。为了做到这一点，首先需要准备一个测试集进行评估。有一个好的测试集可能是最重要的事情，能确保训练的模型以一种可靠的方式在实际环境中使用。创建不具有代表性的测试集，或不使用测试集进行超参数优化，一定会导致模型在实际场景下的应用时表现欠佳。现在创建一个基于随机拆分的简单测试集：

  1. # in Python
  2. train，test = preparedDF.randomSplit([0.7，0.3])

现在已经将数据转换为正确的格式，也创建了有用的特征，终于到了使用模型的时候了。在本例子中将使用一种称为逻辑回归（logistic regression）的分类算法，创建一个 LogisticRegression 的实例，使用默认配置和超参数，然后设置标签列和特征列。这里设置的 label 和 features 列，实际上是 Spark MLlib 估计器默认使用的标签，在后面的内容里会省略它们：

  1. from pyspark.ml.classification import LogisticRegression
  2. lr = LogisticRegression(labelCol="label",featuresCol="features")

实际去训练这个模型之前，先检查一下参数，这也是提示每个特定模型所有可用参数选项的好方法：

  1. # in Python
  2. print lr.explainParams()

虽然输出太大无法在此处重现，但它显示了对 Spark 实现逻辑回归的所有参数的解释，explainParams 方法在 MLlib 可用的所有算法中都有实现。在实例化未经过训练的算法后，该把数据输入进去了，在本例中它将返回 LogisticRegressionModel：

  1. # in Python
  2. fittedLR = lr.fit(train)

这段代码将启动一个 Spark job 来训练模型。与以前的普通 transformation 懒执行过程相反，机器学习模型的拟合是即刻（eager**）实现的。完成后可以使用模型进行预测，从逻辑上说这就是从特征到标签的转换，可以使用 `transform()` 方法进行预测，例如可以 transform 训练集，看看模型对训练数据会输出什么标签，并将其与真实标签进行比较**，这又是可以操纵的另一个 DataFrame，用下面的代码段来执行该预测：

  1. fittedLR.transform(train).select("label", "prediction").show()

结果是：

  1. +-----+----------+
  2. |label|prediction|
  3. +-----+----------+
  4. | 0。0| 0。0|
  5. …
  6. | 0。0| 0。0|
  7. +-----+----------+

下一步将是手动评估此模型，并计算性能指标，如真正率（true positive rate）、假负率（false negative rate）等。然后，可以尝试一组不同的参数以查看这些性能指标是否更好， Spark 可以帮助避免手动尝试不同的模型和性能指标，它允许将 Spark job 指定为包含所有转换的声明式流水线，以及允许调整超参数。

超参数（Hyperparameter**）**是影响训练过程的配置参数，诸如模型结构和正则化（regularization），它们在训练开始之前被设置，例如逻辑回归有一个超参数（正则化参数），它决定了数据在训练阶段执行什么程度的正则化（正则化是一种避免模型过拟合的技术）。后面将会看到，以不同的超参数值（例如不同的正则化参数值）来启动 pipeline job，以便比较同一模型不同版本的效果。

如果执行大量的转换，则编写所有步骤并跟踪 DataFrame 的过程最终会非常繁琐，这就是 Spark 包含 Pipeline 概念的原因所在。pipeline 允许设置相关转换的数据流，以估计器结束，估计器可以根据用户指定自动调整，最后得到一个优化后的模型，即可以直接使用。如下图所示：

image.png

请注意，转换器或模型的实例在不同 pipeline 上不会被重用，在创建另一个 pipeline 之前，始终要创建模型的新实例。为了确保不会过拟合，这里将创建一个测试集并根据验证集调整超参数（注意这里基于原始数据集创建此验证集，而不是前面提到的 preparedDF）：

  1. # in Python
  2. train, test = df.randomSplit([0.7, 0.3])

配置完成后，需要创建 pipeline 中的基本 stage，一个 stage 仅仅代表转换器或估计器。在该例子中将有两个估计器，创建 RFomula 将首先分析数据以了解输入特征的类型，然后转换它们以创建新的特征，随后创建 LogisticRegression 对象，它是通过训练来生成模型的算法：

  1. # in Python
  2. rForm = RFormula()
  3. lr = LogisticRegression().setLabelCol("label").setFeaturesCol("features")

现在，不是去手动使用转换然后调整模型，而是只需在整个 pipeline 的 stage 创建它们，如下面的代码片段所示：

  1. from pyspark.ml import Pipeline
  2. stages = [rForm, lr]
  3. pipeline = Pipeline().setStages(stages)

现在已经设定好了逻辑 pipeline，下一步是训练。在这个例子中不只训练一个模型，将通过指定不同的超参数组合来让 Spark 训练多个不同的模型。然后将使用评估器选择最佳模型，并将它作出的预测与验证数据进行比较。可以测试整个 pipeline 上各种不同的超参数组合，甚至当使用 RFormula 操作原始数据的时候也是如此。下面代码演示了如何执行此操作：

  1. from pyspark.ml.tuning import ParamGridBuilder
  2. params = ParamGridBuilder()\
  3.   .addGrid(rForm.formula, [
  4.     "lab ~ . + color:value1",
  5.     "lab ~ . + color:value1 + color:value2"])\
  6.   .addGrid(lr.elasticNetParam, [0.0, 0.5, 1.0])\
  7.   .addGrid(lr.regParam, [0.1, 2.0])\
  8.   .build()

在当前的参数设置中，有三个超参数与默认值不同：

  • 两个不同版本的 Rformula。
  • 对于 ElasticNet 参数有三个不同的选择。
  • 对于正则化参数有两种不同的选择。

这给了总共 12 种不同的参数组合，这意味着将训练 12 个不同版本的逻辑回归模型。现在已建立了各种参数组合，是指定评估过程的时候了。评估器将自动和客观地把多个模型基于同一评估指标进行比较，在这个例子中使用 BinaryClassificationEvaluator，它有许多潜在的评估指标，并将使用 areaUnderROC，它是受试者工作特征曲线的总面积，这是评估分类任务效果的常用评估指标：

  1. from pyspark.ml.evaluation import BinaryClassificationEvaluator
  2. evaluator = BinaryClassificationEvaluator()\
  3.   .setMetricName("areaUnderROC")\
  4.   .setRawPredictionCol("prediction")\
  5.   .setLabelCol("label")

现在有了指定数据应该如何转换的 pipeline，将对逻辑回归模型执行模型选择，尝试不同的超参数，利用 areaUnderROC 指标比较性能来评估其效果。机器学习中在验证集（而非测试集）上对比超参数效果可以很好地避免过拟合问题，出于这个原因不用之前创建的测试集来调整这些参数。Spark 提供了自动调整超参数的两种选择，可以使用 TrainValidationSplit 简单地将数据任意随机分成两个不同的组，或使用 CrossValidator 通过将数据集分割成不重叠的、随机分配的 k 个折叠来执行 K- 折叠交叉验证（K-fold cross-validation）：

  1. from pyspark.ml.tuning import TrainValidationSplit
  2. tvs = TrainValidationSplit()\
  3.   .setTrainRatio(0.75)\
  4.   .setEstimatorParamMaps(params)\
  5.   .setEstimator(pipeline)\
  6.   .setEvaluator(evaluator)

下面运行构建的 pipeline，运行该 pipeline 将对每一个模型版本都进行测试。注意 tvsFitted 的类型是 TrainValidationSplitModel，当拟合一个模型会输出一个 “model” 类型：

  1. # in Python
  2. tvsFitted = tvs.fit(train)

可以在测试集上评估它的效果：

  1. evaluator.evaluate(tvsFitted.transform(test)) // 0.9166666666666667

也可以查看模型的训练信息总结。要做到这一点要从 pipeline 中提取它，将它转换为适当的类型并打印结果。下面是查看结果的方法：

  1. // in Scala
  2. import org.apache.spark.ml.PipelineModel
  3. import org.apache.spark.ml.classification.LogisticRegressionModel
  4. val trainedPipeline = tvsFitted.bestModel.asInstanceOf[PipelineModel]
  5. val TrainedLR = trainedPipeline.stages(1).asInstanceOf[LogisticRegressionModel]
  6. val summaryLR = TrainedLR.summary
  7. summaryLR.objectiveHistory // 0.6751425885789243，0.5543659647777687，0.473776...

此处显示了算法在训练中每次迭代的训练效果。可以观察到算法正逼近最佳模型，通常是在开始时评价指标会有较大的变化，但随着时间的推移，这些值应该变得越来越小。现在训练好了这个模型，可以将它保存到磁盘上，以便在以后的预测中使用它：

  1. tvsFitted.write.overwrite().save("/tmp/modelLocation")

在写出模型之后，可以将其加载到另一个 Spark 程序中进行预测，为此需要使用特定算法的 “model” 版本来从磁盘上来加载保存的模型。例如如果使用 CrossValidator，则必须在持久化版本中读取 CrossValidatorModel，如果要手动使用 LogisticRegression，则必须使用 LogisticRegressionModel。在当前例子中使用 TrainValidationSplit，它输出 TrainValidationSplitModel：

  1. // in Scala
  2. import org.apache.spark.ml.tuning.TrainValidationSplitModel
  3. val model = TrainValidationSplitModel.load("/tmp/modelLocation")
  4. model.transform(test)

1.5.4 MLlib 常见的工作流程

在 Spark 中有几种不同的部署模式，用于实际应用机器学习模型，下图展示了常见的工作流程：

image.png

流程：

  1. 离线训练机器学习 (ML) 模型，然后向其提供离线数据。离线数据指的是存储下来用于分析的数据，而不是需要快速得到应答的实时数据。Spark 很适合这种部署方式。
  2. 离线训练模型，然后把训练结果放到一个数据库（通常是一个 KV 存储）中。这非常适用于推荐系统，但不适用于像分类或回归这样的应用，因为这些应用不是为某一用户查询返回一个值，而是必须基于输入计算输出值。
  3. 离线训练模型，持久化模型到磁盘上用于之后提供服务。如果使用 Spark 作为服务模块，这将不是一个低延迟的解决方案，因为 Spark job 启动很耗时，即使不是在集群上运行。此外它不能很好的并行化，所以很可能不得不使用负载均衡器，集成多个模型副本提供服务，并集成一些自己的 REST API。这个问题有一些有趣的潜在解决方案，但当前还没有标准化的方法。
  4. 手动（或通过其他一些软件）将分布式模型转换成可以在一台机器上更快运行的模型。当不对 Spark 中的原始数据做很多操作时这种方法很不错，但可能很难维护和更新。当前也有几种解决方案，例如 MLlib 可以将某些模型导出为 PMML（一种通用的模型交换格式）文件。
  5. 在线训练算法并在线使用。可以和 Structured Streaming 一起使用，但是对于某些模型来说会很复杂。