value toDF 不是 org.apache.spark.rdd.RDD[(K， V)] 的成员

Question

注意：以下问题答案中给出的建议不起作用 value toDF 不是 org.apache.spark.rdd.RDD 的成员 value toDF 不是 org.apache.spark.rdd.RDD[Weather] 的成员

我正在尝试编写一个通用函数，该函数仅保留给定数据集中每个键的前 k 值：

下面是代码：

def topKReduceByKey[K:ClassTag,V:Ordering](ds: Dataset[(K, V)], k: Int): Dataset[(K, V)] = {
import sqlContext.implicits._
ds
.rdd
.map(tuple => (tuple._1, Seq(tuple._2)))
.reduceByKey((x, y) => (x ++ y).sorted(Ordering[V].reverse).take(k))
.flatMap(tuple => tuple._2.map(v => (tuple._1, v)))
.toDF("key", "value")
.as[(K, V)]
}

在运行它时，我收到以下错误消息：

Error:(43, 8) value toDF is not a member of org.apache.spark.rdd.RDD[(K, V)]
possible cause: maybe a semicolon is missing before `value toDF'?
.toDF("key", "value")

谁能帮我了解这里出了什么问题？

Answer 1

有多种方法可以做到这一点(分组依据、分区依据、迭代部分(，但只有在您想要自定义分区时才应该使用 RDD，对于其他任何内容，您应该保留数据帧或数据集。

我将使用数据帧提供一个 Python 版本。这应该是 Scala with Datasets 的一个很好的起始示例(API 是相同的(。

def topKByColumn(df, group_column, ordering_column, k):
window = Window.partitionBy(df[group_column]).orderBy(df[ordering_column].desc())
top_k = df.withColumn('rank', row_number().over(window))
top_k = top_per[top_k.rank <= k]
return top_k