一、实验目的

掌握使用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。