Spark中那些常用的特征处理操作

importorg.apache.spark.ml.feature.{CountVectorizer, CountVectorizerModel}
val df =spark.createDataFrame(Seq(
      (0, Array("a","e","a","d","b")),
      (1, Array("a","c","b","d","c","f","a","b")),
      (2, Array("a","f"))
)).toDF("id", "words")   
var cv_model = newCountVectorizer().setInputCol("words").setOutputCol("features").setVocabSize(10).setMinDF(2).fit(df)
val cv1 = cv_model.transform(df)
cv1.show(false)

import org.apache.spark.ml.feature.{HashingTF,IDF, Tokenizer}
val wordsData =spark.createDataFrame(Seq(
"传奇 游戏 战士".split(" "),
"苹果 梨 香蕉".split(" "),
"苹果 手机 流畅".split(" ")
    ).map(Tuple1.apply)).toDF("words")
wordsData.show(false)   
// step1 hashingTF
val hashingTF = newHashingTF().setInputCol("words").setOutputCol("rawFeatures").setNumFeatures(2000)
val featurizedData =hashingTF.transform(wordsData)
// step2 计算IDF
val idf = new IDF().setInputCol("rawFeatures").setOutputCol("features")
val idfModel =idf.fit(featurizedData)
val rescaledData =idfModel.transform(featurizedData)
rescaledData.select("words","features").show(false)

?importorg.apache.spark.ml.feature.Word2Vec
val documentDF =spark.createDataFrame(Seq(
      "传奇 游戏 战士".split(" "),
"苹果 梨 香蕉".split(" "),
"传奇 游戏 种类多".split(" "),
"苹果 手机 流畅".split(" ")
    ).map(Tuple1.apply)).toDF("text")
val word2Vec = newWord2Vec().setInputCol("text").setOutputCol("result").setVectorSize(10).setMinCount(2)
val model =word2Vec.fit(documentDF)
val result =model.transform(documentDF)
result.show(false)

?importorg.apache.spark.ml.feature.Bucketizer
val data = Array(-8.0, -0.5, -0.3,0.0, 0.2, 9.0)
val splits = Array(Double.NegativeInfinity,-0.5, 0.0, 0.5, Double.PositiveInfinity)
val dataFrame =spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")
val bucketizer = newBucketizer().setInputCol("features").setOutputCol("bucketedFeatures").setSplits(splits)
bucketizer.transform(dataFrame).show(false)

?importorg.apache.spark.ml.feature.QuantileDiscretizer
val data = Array((0, 18.0), (1, 19.0),(2, 8.0), (3, 5.0), (4, 2.2))
var df =spark.createDataFrame(data).toDF("id", "hour")
val discretizer = newQuantileDiscretizer().setInputCol("hour").setOutputCol("result").setNumBuckets(3)
val result =discretizer.fit(df).transform(df)
result.show()

importorg.apache.spark.ml.feature.StringIndexer
val df = spark.createDataFrame(
 Seq((0, "a"), (1, "b"), (2, "c"), (3, "a"),(4, "a"), (5, "c"))
).toDF("id", "category")
val indexer = newStringIndexer().setInputCol("category").setOutputCol("categoryIndex")
val indexed =indexer.fit(df).transform(df)
indexed.show(false)

?importorg.apache.spark.ml.feature.Normalizer
importorg.apache.spark.ml.linalg.{Vector,Vectors}
val data=Seq(Vectors.dense(-1,1,1,8,56),Vectors.dense(-1,3,-1,-9,88),Vectors.dense(0,5,1,10,96),Vectors.dense(0,5,1,11,589),Vectors.dense(0,5,1,11,688))
valdf=spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")   
val normalizer = newNormalizer().setInputCol("features").setOutputCol("normFeatures").setP(1.0)
normalizer.transform(df).show(false)

?importorg.apache.spark.ml.feature.StandardScaler
importorg.apache.spark.ml.linalg.{Vector,Vectors}
val dataFrame =spark.createDataFrame(Seq(
(0, Vectors.dense(1.0, 0.5, -1.0)),(1,Vectors.dense(2.0, 1.0, 1.0)),
(2, Vectors.dense(4.0, 10.0, 2.0)))).toDF("id","features")
val scaler = newStandardScaler().setInputCol("features")
.setOutputCol("scaledFeatures").setWithStd(true).setWithMean(false)
val scalerModel =scaler.fit(dataFrame)
val scaledData =scalerModel.transform(dataFrame)
scaledData.show(false)

importorg.apache.spark.ml.feature.VectorIndexer
importorg.apache.spark.ml.linalg.Vectors
val data=Seq(Vectors.dense(-1,1,1,8,56),
             Vectors.dense(-1,3,-1,-9,88),
             Vectors.dense(0,5,1,10,96),
             Vectors.dense(0,5,1,11,589))
valdf=spark.createDataFrame(data.map(Tuple1.apply)).toDF("features")   
val indexer = newVectorIndexer().setInputCol("features").setOutputCol("indexed").setMaxCategories(3)
val indexerModel =indexer.fit(df)
indexerModel.transform(df).show(false)

?import org.apache.spark.ml.feature.SQLTransformer
val df = spark.createDataFrame(
 Seq((0, 1.0, 3.0), (2, 2.0, 5.0))).toDF("id", "v1", "v2")
val sqlTrans = newSQLTransformer().setStatement(
"SELECT *, (v1 + v2) AS v3, (v1 * v2) AS v4 FROM __THIS__")
sqlTrans.transform(df).show()

importorg.apache.spark.ml.feature.ChiSqSelector
importorg.apache.spark.ml.feature.VectorIndexer
val data = Seq( (7,Vectors.dense(0.0, 0.0, 18.0, 1.0), 1.0),
 (8, Vectors.dense(0.0, 1.0, 12.0, 0.0), 0.0), (9, Vectors.dense(1.0, 0.0,15.0, 0.2), 0.0))
val df =spark.createDataset(data).toDF("id", "features", "clicked")
val selector = newChiSqSelector().setNumTopFeatures(2).setFeaturesCol("features").setLabelCol("clicked").setOutputCol("selectedFeatures")
val result =selector.fit(df).transform(df)
result.show(false)