一、实验目的
掌握使用Spark机器学习管道创建小型机器学习工作流。
二、实验内容
1、构建一个机器学习管道，应用LogisticRegression算法，预测一行文本中是否出现了”spark”这个单词。
三、实验原理
Spark ML有一个名为Pipeline的类，它被设计用来管理一系列的阶段，每一个阶段都由PipelineStage来表示。一个PipelineStage既可以是transformer，也可以是estimator。抽象Pipeline是一种estimator。管道以指定的顺序连接多个transformers和estimators，形成机器学习工作流。从概念上讲，它将机器学习工作流中的数据预处理、特征提取和模型训练步骤链接在一起。
管道由一系列阶段组成，每个阶段都是一个Transformer或一个Estimator。它按照指定的顺序运行这些阶段。
下图描述了一个使用管道创建一个小型工作流。
四、实验环境
硬件：x86_64 ubuntu 16.04服务器
软件：JDK 1.8，Spark-2.3.2，Hadoop-2.7.3，zeppelin-0.8.1
五、实验步骤
5.1 启动Spark集群和Zeppelin服务器。
在终端窗口下，输入以下命令，分别启动Spark集群和Zeppelin服务器：
1. $ cd /opt/spark
2. $ ./sbin/start-all.sh
3. $ zeppelin-daemon.sh start
然后使用jps命令查看启动的进程，确保Spark集群和Zeppelin服务器已经正确启动。
2、创建notebook。启动浏览器，访问”http://localhost:9090“, 打开zeppelin notebook首页，点击”Create new note”链接，创建一个新的笔记本。如下图所示：
5.2 使用管道创建一个小型工作流。
这个示例中，管道由两个transformers和一个estimator组成。当调用Pipeline.fit()函数时，包含原始文本的输入DataFrame将被传递给Tokenizer transformer，其输出将被传递到HashingTF transformer，它将单词转换为特征。该Pipeline认识到LogisticRegression是一个estimator，因此它将调用fit函数和计算特征来产生一个LogisticRegressionModel。
1、导入所需的包。在zeppelin中输入以下代码：
1. import org.apache.spark.ml.{Pipeline, PipelineModel}
2. import org.apache.spark.ml.classification.LogisticRegression
3. import org.apache.spark.ml.feature.{HashingTF, Tokenizer}
同时按下”【Shift + Enter】”键，执行以上代码。
2、构造一个DataFrame。在zeppelin中输入以下代码：
1. val text_data = spark.createDataFrame(Seq(
2. (1, "Spark is a unified data analytics engine", 0.0),
3. (2, "Spark is cool and it is fun to work with Spark", 0.0),
4. (3, "There is a lot of exciting sessions at upcoming Spark summit", 0.0),
5. (4, "signup to win a million dollars", 0.0) )
6. ).toDF("id", "line", "label")
同时按下”【Shift + Enter】”键，执行以上代码。
3、构造第一个阶段transformer。在zeppelin中输入以下代码：
1. val tokenizer = new Tokenizer().setInputCol("line").setOutputCol("words")
同时按下”【Shift + Enter】”键，执行以上代码。
4、构造第二个阶段transformer(第一个阶段的输出作为第二个阶段的输入)。在zeppelin中输入以下代码：
1. val hashingTF = new HashingTF().setInputCol(tokenizer.getOutputCol)
2. .setOutputCol("features")
3. .setNumFeatures(4096)
同时按下”【Shift + Enter】”键，执行以上代码。
5、构造第三个阶段estimator，代表逻辑回归算法实现。在zeppelin中输入以下代码：
1. val logisticReg = new LogisticRegression().setMaxIter(5).setRegParam(0.01)
同时按下”【Shift + Enter】”键，执行以上代码。
6、构造一个管道，由以上三个阶段组成。在zeppelin中输入以下代码：
1. val pipeline = new Pipeline().setStages(Array(tokenizer, hashingTF, logisticReg))
同时按下”【Shift + Enter】”键，执行以上代码。
7、触发各阶段的顺序执行。在zeppelin中输入以下代码：
1. val logisticRegModel = pipeline.fit(text_data)
同时按下”【Shift + Enter】”键，执行以上代码。
8、使用学习到的模型对数据进行转换。在zeppelin中输入以下代码：
1. logisticRegModel.transform(text_data).show
同时按下”【Shift + Enter】”键，执行以上代码。输出结果如下：
结束语
Pipeline的fit方法调用每个Transformer的transform方法和每个Estimator的fit方法，与创建管道时指定的顺序相同。每个Transformer接受一个DataFrame作为输入，并返回一个新的DataFrame，它将成为管道中下一阶段的输入。如果一个阶段是一个Estimator，则调用它的fit方法来训练一个模型。返回的模型是一个Transformer，用于将前一阶段的输出转换为下一阶段的输入。管道本身也是一个Estimator，其fit方法返回一个PipelineModel，它是一个Transformer。
