Spark Dataframes - 关键减少

Posted 2023-04-15

【中文标题】Spark Dataframes - 关键减少

【英文标题】：Spark Dataframes- Reducing By Key

【发布时间】：2016-12-20 07:20:26

【问题描述】：

假设我有一个像这样的数据结构，其中 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 并且我不清楚我需要编写什么聚合函数来实现我想要的 - 我见过的所有示例聚合似乎都专注于只返回一个被聚合的列而不是整个行。

是否可以使用数据框来实现这一点？

【参考方案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.*")

【讨论】：

是的，效果很好，谢谢！顺便说一句，我认为你的答案有一个轻微的错字 - agg 之后的句点应该是一个括号！

您能解释一下为什么会这样吗？最大值是否仅应用于第一列？

@user238607 max 使用列的排序。结构（或元组）的隐式排序是按第一个元素的排序。

工作正常，同意 TS，有一个错字 - 应该有 (agg & max 之间

【参考方案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