scala - 无法在 Spark 2.0 中的数据集 [(scala.Long, org.apache.spark.mllib.linalg.Vector)] 上运行 LDA

Question

我正在关注 this关于 LDA 示例的教程视频，我遇到了以下问题:

<console>:37: error: overloaded method value run with alternatives:
  (documents: org.apache.spark.api.java.JavaPairRDD[java.lang.Long,org.apache.spark.mllib.linalg.Vector])org.apache.spark.mllib.clustering.LDAModel <and>
  (documents: org.apache.spark.rdd.RDD[(scala.Long, org.apache.spark.mllib.linalg.Vector)])org.apache.spark.mllib.clustering.LDAModel
  cannot be applied to (org.apache.spark.sql.Dataset[(scala.Long, org.apache.spark.mllib.linalg.Vector)])
     val model = run(lda_countVector)
                                   ^

所以我想将这个 DF 转换为 RDD，但它总是被分配为数据集。任何人都可以调查这个问题吗？

// Convert DF to RDD
import org.apache.spark.mllib.linalg.Vector
val lda_countVector = countVectors.map { case Row(id: Long, countVector: Vector) => (id, countVector) }
// import org.apache.spark.mllib.linalg.Vector
// lda_countVector: org.apache.spark.sql.Dataset[(Long, org.apache.spark.mllib.linalg.Vector)] = [_1: bigint, _2: vector]

Answer 1

Spark API 在 1.x 和 2.x 分支之间发生了变化。特别是 DataFrame.map 返回 Dataset 而不是 RDD 因此结果与基于旧 MLlib RDD 的不兼容应用程序接口(interface)。您应该首先将数据转换为 RDD，如下所示:

import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.sql.Row
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.clustering.{DistributedLDAModel, LDA}

val a = Vectors.dense(Array(1.0, 2.0, 3.0))
val b = Vectors.dense(Array(3.0, 4.0, 5.0))
val df = Seq((1L ,a), (2L, b), (2L, a)).toDF

val ldaDF = df.rdd.map { 
  case Row(id: Long, countVector: Vector) => (id, countVector) 
} 

val model = new LDA().setK(3).run(ldaDF)

或者您可以转换为类型化数据集，然后再转换为 RDD:

val model = new LDA().setK(3).run(df.as[(Long, Vector)].rdd)