Spark 高级分析:第十三章附录：即将推出的MLLIB管道API

问题导读

1.什么是Spark MLLIB？
2.真正的生产机器学习部署涉及哪些任务？
3.如何使用Spark MLlib进行文本分类？



上一篇
Spark 高级分析:第十二章第4-6节 Spark和数据科学家的工作流程
http://www.aboutyun.com/forum.php?mod=viewthread&tid=26821


Spark项目进展迅速。当我们在2014年8月开始编写时，版本1.1.0即将发布。这本书将于2015年1月出版，Spark 1.2.0也正在热销。仅这个小版本就包含了近800个修复和改进。

该项目小心地维护小版本中稳定API的二进制和源代码兼容性，并且大多数MLLIB被认为是稳定的。因此，本书中的示例应该继续使用spark 1.3.0和未来的1.x版本；这些实现不会出现在任何地方。然而，新版本通常会添加或更改实验性的或仅限于开发人员的API，这些API仍在不断发展。

当然，spark mllib在这些章节中有着突出的特点，一本涵盖spark 1.2.0的书如果没有提到MLlib的一个重要的新方向，就不完整了，这在spark 1.2.0中部分地作为实验API出现：管道API。

这本书在官方上只有一个月左右的历史，可能会有变化，而且还没有接近完成，因此还不可能围绕这本书进行创作。然而，这是值得了解的，已经看到了MLLIB今天提供的。

本附录将快速介绍新的管道API，即Spark-3530在Spark项目问题跟踪程序中讨论的工作结果。

第1节不仅仅是建模

在目的和范围上，当前的MLLIB类似于其他机器学习库。它提供了一个机器学习算法的实现，只是核心实现。例如，每个都将预处理的输入作为labeledpoint或rating对象的RDD，并返回结果模型的一些表示。仅此而已。这是非常有用的，但是解决一个真实的机器学习问题需要的不仅仅是运行一个算法。

你可能已经注意到，在本书的每一章中，大多数源代码的存在都是为了从原始输入准备特性、转换特性以及以某种方式评估模型。调用MLLIB算法只是中间的一个小而简单的部分。
这些额外的任务对于任何机器学习问题都是常见的。实际上，真正的生产机器学习部署可能涉及更多任务：
1。将原始数据解析为功能
2。将功能转换为其他功能
3。建立模型
4。评估模型
5。调整模型超参数
6。持续重建和部署模型
7。实时更新模型
8。实时回答来自模型的查询
这样看来，MLLIB只提供了一小部分：3。新的PipelinesAPI开始扩展MLLIB，使其成为处理任务1到5的框架。这正是我们在整本书中必须以不同方式手工完成的任务。其余部分很重要，但可能超出MLLIB的范围。这些方面可以通过Spark Streaming、JPMML、RESTAPI、Apache Kafka等工具的组合来实现。

第2节管道API

新的Pipelines API封装了上面机器学习任务的简单、整洁的视图：在每个阶段，数据都被转换成其他数据，并最终转换成一个模型，这个模型本身也是一个只从其他数据（输入）创建数据（预测）的实体。

这里的数据总是由从Spark Sql（org.apache.spark.sql.schemardd类）借用的专用RDD表示。顾名思义，它包含类似表的数据，其中每个元素都是一行。每行有相同的“列”，其模式是已知的，包括名称、类型等。

这使得类似SQL的操作能够方便地转换、投影、筛选和联接这些数据。与Spark的其他API一起，这主要解决了上面的任务1。

更重要的是，模式信息的存在意味着机器学习算法能够更准确、更自动地区分数字和分类特征。输入不再只是一个双值数组，调用者负责通信，而实际上是分类的。

新管道API的其余部分，或者至少已经作为实验API发布供预览的部分，都位于org.apache.spark.ml包下——与org.apache.spark.mllib包中当前稳定的API相比。

Transformer抽象表示可以将数据转换为其他数据的逻辑——一个SchemaRdd转换为另一个SchemaRdd。一个估计量表示可以从SchemaRdd构建机器学习模型或模型的逻辑。模型本身就是一个转换的。

org.apache.spark.ml.feature包含一些有用的实现，比如在TF-IDF中计算术语频率的HashingTF，或者简单解析的标记器。这样，新的API有助于支持上面的任务2。

然后，管道抽象表示一系列Transformer和Estimator对象，这些对象可以按顺序应用于输入模式，以便输出模型。因此，pipeline本身就是一个估计量，因为它生成一个模型！

这种设计允许一些有趣的组合。因为管道可能包含一个估计量，这意味着它可以在内部构建一个模型，然后将该模型用作transformer。也就是说，管道可以在内部构建和使用算法的预测，作为更大流量的一部分。实际上，这也意味着管道中可以包含其他管道实例。

为了回答任务3，在这个新的实验API（org.apache.spark.ml.classification.logisticregression）中，已经有了至少一个实际建模算法的简单实现。虽然可以将现有的org.apache.spark.mllib实现包装为一个估计量，但是新的API已经为我们提供了一个重写的逻辑回归实现，例如评估器抽象支持对模型预测的评估。反过来，它被用于org.apache.spark.ml.tuning中的CrossValidator类中，以从SchemaRdd创建和评估许多模型实例，因此，它也是一个估计量。支持org.apache.spark.ml.params中的API定义用于交叉验证程序的超参数和网格搜索参数。这些包有助于完成任务4和5，然后评估和调整模型，作为更大管道的一部分。

第3节文本分类示例演练

spark examples模块在org.apache.spark.examples.ml.simpleTextClassificationPipeline类中包含一个新API的简单示例。其作用如图13-1所示：

输入是表示文档的对象，具有ID、文本和分数（标签）。尽管training 不是一个SchemaRdd，但稍后将隐式转换它。
[mw_shl_code=scala,true]val training = sparkContext.parallelize(Seq(
LabeledDocument(0L, "a b c d e spark", 1.0),
LabeledDocument(1L, "b d", 0.0),
LabeledDocument(2L, "spark f g h", 1.0),
LabeledDocument(3L, "hadoop mapreduce", 0.0)))[/mw_shl_code]
管道应用两个转换实现。首先，标记器通过空格将文本分隔成单词。然后，hashingf计算每个单词的词频。最后，逻辑回归创建了一个分类器，使用这些术语频率作为输入特征。
[mw_shl_code=scala,true]val tokenizer = new Tokenizer().
setInputCol("text").
setOutputCol("words")
val hashingTF = new HashingTF().
setNumFeatures(1000).
setInputCol(tokenizer.getOutputCol).
setOutputCol("features")
val lr = new LogisticRegression().
setMaxIter(10).
setRegParam(0.01)[/mw_shl_code]
这些操作被合并到一个管道中，该管道实际上从训练输入创建了一个模型。
[mw_shl_code=scala,true]val pipeline = new Pipeline().
setStages(Array(tokenizer, hashingTF, lr))
val model = pipeline.fit(training)[/mw_shl_code]
最后，该模型可用于新文档的分类。注意，模型实际上是包含上面所有转换逻辑的管道，而不仅仅是对分类器模型的调用。
[mw_shl_code=scala,true]val test = sparkContext.parallelize(Seq(
Document(4L, "spark i j k"),
Document(5L, "l m n"),
Document(6L, "mapreduce spark"),
Document(7L, "apache hadoop")))
model.transform(test).
select('id, 'text, 'score, 'prediction).
collect().
foreach(println)[/mw_shl_code]
与目前在MLib周围实现相同功能所必需的手写代码相比，整个管道的代码更简单、更有组织、更可重用。

最新经典文章，欢迎关注公众号

感谢分享