1.  $ cd /opt/spark
2.  $ ./sbin/start-all.sh
3.  $ zeppelin-daemon.sh start

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)

1.  // 使用StringIndexer estimator来对电影类型进行编码
2.  import org.apache.spark.ml.feature.StringIndexer
3.      　
4.  // 构造一个DataFrame
5.  val movie_data = spark.createDataFrame(
6.                              Seq((1, "Comedy"),
7.                                  (2, "Action"),
8.                                  (3, "Comedy"),
9.                                  (4, "Horror"),
10.                                 (5, "Action"),
11.                                 (6, "Comedy"))
12.                         ).toDF("id", "genre")
13.     　
14. // StringIndexer estimator
15. val movieIndexer = new StringIndexer().setInputCol("genre").setOutputCol("genreIdx")
16.     　
17. // 首先拟合数据
18. val movieIndexModel = movieIndexer.fit(movie_data)
19.     　
20. // 使用返回的transformer来转换该数据
21. val indexedMovie = movieIndexModel.transform(movie_data)
22.     　
23. // 查看结果
24. indexedMovie.orderBy("genreIdx").show()

1.  // OneHotEncoderEstimator estimator消费StringIndexer estimator的输出
2.  import org.apache.spark.ml.feature.OneHotEncoderEstimator
3.      　
4.  // 输入列genreIdx是之前示例中StringIndex的输出列
5.  val oneHotEncoderEst = new OneHotEncoderEstimator().setInputCols(Array("genreIdx"))
6.  .setOutputCols(Array("genreIdxVector"))
7.      　
8.  // 指使indexedMovie数据（在上一个示例中产生的）
9.  val oneHotEncoderModel = oneHotEncoderEst.fit(indexedMovie)
10. val oneHotEncoderVect = oneHotEncoderModel.transform(indexedMovie)
11.     　
12. // 显示
13. oneHotEncoderVect .orderBy("genre").show()

1.  // 使用MinMaxScaler estimator来重新调节特征
2.  import org.apache.spark.ml.feature.MinMaxScaler
3.  import org.apache.spark.ml.linalg.Vectors
4.      　
5.  // 构造DataFrame
6.  val employee_data = spark.createDataFrame(
7.                              Seq((1, Vectors.dense(125400, 5.3)),
8.                                  (2, Vectors.dense(179100, 6.9)),
9.                                  (3, Vectors.dense(154770, 5.2)),
10.                                 (4, Vectors.dense(199650, 4.11)))
11.                         ).toDF("empId", "features")
12.     　
13. // MinMaxScaler estimator
14. val minMaxScaler = new MinMaxScaler().setMin(0.0)
15.                                      .setMax(5.0)
16.                                      .setInputCol("features")
17.                                      .setOutputCol("scaledFeatures")
18.     　
19. // 拟合数据，建立模型
20. val scalerModel = minMaxScaler.fit(employee_data)
21.     　
22. // 使用学习到的模型对数据集进行转换
23. val scaledData = scalerModel.transform(employee_data)
24.     　
25. // 输出特征缩放到的范围
26. println(s"特征缩放到的范围: [${minMaxScaler.getMin},${minMaxScaler.getMax}]")
27.     　
28. // 显示结果
29. scaledData.select("features", "scaledFeatures").show(false)

1.  // 使用StandardScaler estimator标准化围绕均值0的特征
2.  import org.apache.spark.ml.feature.StandardScaler
3.  import org.apache.spark.ml.linalg.Vectors
4.      　
5.  // 构造DataFrame
6.  val employee_data = spark.createDataFrame(Seq(
7.                              (1, Vectors.dense(125400, 5.3)),
8.                              (2, Vectors.dense(179100, 6.9)),
9.                              (3, Vectors.dense(154770, 5.2)),
10.                             (4, Vectors.dense(199650, 4.11)))
11.                         ).toDF("empId", "features")
12.     　
13. // 将单位标准偏差设置为true并围绕平均值
14. val standardScaler = new StandardScaler().setWithStd(true)
15.                                          .setWithMean(true)
16.                                          .setInputCol("features")
17.                                          .setOutputCol("scaledFeatures")
18.     　
19. // 拟合数据，建立模型
20. val standardMode = standardScaler.fit(employee_data)
21.     　
22. // 使用学习到的模型对数据集进行转换
23. val standardData = standardMode.transform(employee_data)
24.     　
25. // 显示结果
26. standardData.show(false)