значение toDF не является членом org.apache.spark.rdd.RDD [(K, V)]

Question

Примечание: предложения, приведенные в ответах на следующие вопросы, не работают значение toDF не является членом org.apache.spark.rdd.RDD значение toDF не является членом org.apache.spark.rdd.RDD [Погода]

Я пытаюсь написать универсальную функцию, которая сохраняет только верхнее значение 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 должен иметь место только в том случае, если вы хотите иметь пользовательское разбиение, для всего остального вы должны остаться с Dataframe или Datasets.

Я предоставлю версию Python, используя Dataframes.Это должно быть хорошим начальным примером для Scala с наборами данных (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