scala - Spark数据帧-按键减少

Question

假设我有一个像这样的数据结构，其中ts是一些时间戳

case class Record(ts: Long, id: Int, value: Int)

给定大量的这些记录，我希望最终得到每个ID时间戳都最高的记录。我认为使用RDD api可以完成以下代码:

def findLatest(records: RDD[Record])(implicit spark: SparkSession) = {
  records.keyBy(_.id).reduceByKey{
    (x, y) => if(x.ts > y.ts) x else y
  }.values
}

同样，这是我对数据集的尝试:

def findLatest(records: Dataset[Record])(implicit spark: SparkSession) = {
  records.groupByKey(_.id).mapGroups{
    case(id, records) => {
      records.reduceLeft((x,y) => if (x.ts > y.ts) x else y)
    }
  }
}

我正在尝试找出如何实现与数据帧相似的方法，但无济于事-我意识到我可以使用以下方法进行分组:

records.groupBy($"id")

但这给了我一个RelationGroupedDataSet，而且我不清楚要实现我想要的功能需要编写什么聚合函数-我看到的所有示例聚合似乎都集中在返回仅聚合的单个列而不是整个行。

是否有可能使用数据框来实现？

Answer 1

您可以使用argmax逻辑(请参阅databricks example)

例如，假设您的数据框称为df，并且具有id，val，ts列，您可以执行以下操作:

import org.apache.spark.sql.functions._
val newDF = df.groupBy('id).agg.max(struct('ts, 'val)) as 'tmp).select($"id", $"tmp.*")