Spark机器学习管道 - Estimator

一、实验目的
二、实验内容
三、实验原理
四、实验环境
五、实验步骤
- 5.1 启动Spark集群和Zeppelin服务器。
- 5.2 使用IDF estimator，计算每个单词的重要性。
- 5.3 使用StringIndexer estimator来对电影类型进行编码。
- 5.4 使用OneHotEncoderEstimator estimator将分类值的索引编码为二元向量。
- 5.5 使用MinMaxScaler estimator对数值数据进行规范化。
- 5.6 使用MinMaxScaler estimator对数值数据进行标准化。
结束语

未经许可，禁止以任何形式转载，若要引用，请标注链接地址
全文共计6161字，阅读大概需要3分钟

一、实验目的

掌握Spark机器学习管道中常用Estimator的使用。

二、实验内容

1、使用IDF estimator，计算每个单词的重要性。
2、使用StringIndexer estimator来对电影类型进行编码。
3、使用OneHotEncoderEstimator estimator将分类值的索引编码为二元向量。
4、使用MinMaxScaler estimator对数值数据进行规范化。
5、使用MinMaxScaler estimator对数值数据进行标准化。

三、实验原理

一个Estimator代表了一种机器学习算法，用来在训练数据集上训练或拟合机器学习模型。它实现了一个名为fit的方法，该方法接受一个DataFrame作为参数并返回一个机器学习模型。
Estimator所代表的算法可分为两类，一类是用于机器学习的算法，另一类是用于进行数据转换的算法。
从技术的角度来看，一个estimator有一个名为fit的函数，它在输入列上应用一个算法，结果被封装在一个叫做Model的对象类型中，它是一个Transformer类型。输入列和输出列名称可以在estimator的构造过程中指定。
下图描述了一个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 使用IDF estimator，计算每个单词的重要性。

IDF estimator是用于处理文本的常用的estimators之一。它的名字是inverse document frequency（反转文档频率）的缩写。这个estimator经常在文本被分词和术语频率被计算之后立即使用。这个estimator背后的思想是通过计算它出现的文档数量来计算每个单词的重要性或权重。
在zeppelin中输入以下代码：

1.   // 使用IDF estimator来计算每个单词的权重
2.  import org.apache.spark.ml.feature.Tokenizer
3.  import org.apache.spark.ml.feature.HashingTF
4.  import org.apache.spark.ml.feature.IDF
5.      　
6.  // 构造一个DataFrame，代表一个文档
7.  val text_data = spark.createDataFrame(Seq(
8.                              (1, "Spark is a unified data analytics engine"),
9.                              (2, "Spark is cool and it is fun to work with Spark"),
10.                             (3, "There is a lot of exciting sessions at upcoming Spark summit"),
11.                             (4, "mllib transformer estimator evaluator and pipelines") )
12.                 ).toDF("id", "line")
13.     　
14. // 分析转换器
15. val tokenizer = new Tokenizer().setInputCol("line").setOutputCol("words")
16. val tkResult = tokenizer.transform(text_data)
17.     　
18. // HashingTF转换器
19. val tf = new HashingTF().setInputCol("words").setOutputCol("wordFreqVect").setNumFeatures(4096)
20. val tfResult = tf.transform(tkResult )    // Tokenizer transformer的输出列是HashingTF的输入
21.     　
22. // IDF estimator
23. // HashingTF转换器的输出是IDF estimator的输入
24. val idf = new IDF().setInputCol("wordFreqVect").setOutputCol("features")
25.     　
26. // 因为IDF是一个estimator,所以调用fit函数, 得到一个学习过的模型
27. val idfModel = idf.fit(tfResult)
28.     　
29. // 返回对象是一个模型（Model）, 它是Transformer类型
30. val weightedWords = idfModel.transform(tfResult)
31. // weightedWords.select("label", "features").show(false)
32.  weightedWords.select("features").show(false)