Spark DataFrames-通过键减少

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")

但这给了我一个关系groupedDataSet，我还不清楚我需要写的聚合函数以实现我想要的东西 - 我所看到的所有示例聚合似乎都集中在返回一列，而不是整个列，而不是整个列行。

是否可以使用dataframes？

Answer 1

您可以使用argmax逻辑（请参阅databricks示例）

例如，假设您的数据框称为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.*")

Answer 2

对于数据集，我这样做，在Spark 2.1.1

上进行了测试

final case class AggregateResultModel(id: String,
                                      mtype: String,
                                      healthScore: Int,
                                      mortality: Float,
                                      reimbursement: Float)
.....
.....
// assume that the rawScores are loaded behorehand from json,csv files
val groupedResultSet = rawScores.as[AggregateResultModel].groupByKey( item => (item.id,item.mtype ))
      .reduceGroups( (x,y) => getMinHealthScore(x,y)).map(_._2)

// the binary function used in the reduceGroups
def getMinHealthScore(x : AggregateResultModel, y : AggregateResultModel): AggregateResultModel = {
    // complex logic for deciding between which row to keep
    if (x.healthScore > y.healthScore) { return y }
    else if (x.healthScore < y.healthScore) { return x }
    else {
      if (x.mortality < y.mortality) { return y }
      else if (x.mortality > y.mortality) { return x }
      else  {
        if(x.reimbursement < y.reimbursement)
          return x
        else
          return y
      }
    }
  }