apache-spark - 将 DataFrame 与 MLlib 一起使用

Question

假设我有一个 DataFrame(我从 HDFS 上的 csv 读取)，我想通过 MLlib 训练一些算法。如何将行转换为 LabeledPoints 或在此数据集上使用 MLlib？

Answer 1

基于 RDD 的 Mllib 即将被弃用，因此您应该使用基于 DataFrame 的 Mllib。

通常，这些 MLlib api 的输入是包含 2 列的 DataFrame - 标签和特征。有多种方法可以构建此 DataFrame - 低级 API，例如 org.apache.spark.mllib.linalg.{Vector, Vectors}、org.apache.spark.mllib.regression.LabeledPoint、org.apache.spark.mllib.linalg .{Matrix, Matrices} 等。它们都采用数值作为特征和标签。
可以使用 - org.apache.spark.ml.feature.{Word2Vec, Word2VecModel} 将单词转换为向量。本文档解释更多 - https://spark.apache.org/docs/latest/mllib-data-types.html

创建带有标签和特征的输入数据框后，实例化 MLlib api 并将数据框传递给“fit”函数以获取模型，然后在模型上调用“transform”或“predict”函数以获取结果。

例子 -

培训文件看起来像 -<numeric label> <a string separated by space>

//Build word vector
    val trainingData = spark.read.parquet(<path to training file>)
    val sampleDataDf = trainingData
      .map { r =>
        val s = r.getAs[String]("value").split(" ")
        val label = s.head.toDouble
        val feature = s.tail
        (label, feature)
      }.toDF("lable","feature_words")

    val word2Vec = new Word2Vec()
      .setInputCol("feature_words")
      .setOutputCol("feature_vectors")
      .setMinCount(0)
      .setMaxIter(10)

    //build word2Vector model
    val model = word2Vec.fit(sampleDataDf) 
    //convert training text data to vector and labels
    val wVectors = model.transform(sampleDataDf)

    //train LinearSVM model
    val svmAlgorithm = new LinearSVC()
                    .setFeaturesCol("feature_vectors")
                    .setMaxIter(100)
                    .setLabelCol("lable")
                    .setRegParam(0.01)
                    .setThreshold(0.5)
                    .fit(wVectors) //use word vectors created

    //Predict new data, same format as training data containing words
     val predictionData = spark.read.parquet(<path to prediction file>)

    val pDataDf = predictionData
      .map { r =>
        val s = r.getAs[String]("value").split(" ")
        val label = s.head.toDouble
        val feature = s.tail
        (label, feature)
      }.toDF("lable","feature_words")

    val pVectors = model.transform(pDataDf)
    val predictionlResult = pVectors.map{ r =>
        val s = r.getAs[Seq[String]]("feature_words")
        val v = r.getAs[Vector]("feature_vectors")
        val c = svmAlgorithm.predict(v) // predict using trained SVM

        s"$c ${s.mkString(" ")}"
      }