聚合后获取Spark DataFrame的所有列[重复]

Posted 2023-04-15

【中文标题】聚合后获取Spark DataFrame的所有列[重复]

【英文标题】：Getting all columns of Spark DataFrame after aggregation [duplicate]

【发布时间】：2017-09-26 14:28:21

【问题描述】：

假设我有一个数据框：

+----+----+---+
|  c1|name|qty|
+----+----+---+
|abc1|   a|  1|
|abc2|   a|  0|
|abc3|   b|  3|
|abc4|   b|  2|
+----+----+---+

我只想为每个 name 获取具有最少 qty 的行：

+----+----+---+
|  c1|name|qty|
+----+----+---+
|abc2|   a|  0|
|abc4|   b|  2|
+----+----+---+

我就是这样做的：

df1 = df.groupBy('name').agg(sf.min('qty')).select("min(qty)")
df2 = df1.join(df, df1["min(qty)"] == df["qty"]).drop("min(qty)") // df2 is the result

它正在工作。我想知道它是否可以改进。您如何改进上述解决方案？

【问题讨论】：

Code Review 让工作代码更好

【参考方案1】：

您可以在数据框上使用 reduceBuKey。根据名称减少，然后选择下键。我假设 df 与具有

的数据集相关联

case class (c1:String, name:String, qty:Integer)

val mappedPairRdd =
    df.map(row ⇒ (row.name, (row.c1, row.name, row.qty))).rdd;
  val reduceByKeyRDD = mappedPairRdd
    .reduceByKey((row1, row2) ⇒ 
      if (row1._3 > row2._3) 
        row2
       else 
        row1
      
    )
    .values;