使用无法直接从GroupedData类调用的方法聚合多个列（如“last（）”）并将它们重命名为原始名称[duplicate]

Posted 2021-04-18

这个问题在这里已有答案：

• Spark SQL: apply aggregate functions to a list of column 2回答

假设我们有以下DF

scala> df.show
+---+----+----+----+-------------------+---+
| id|name| cnt| amt|                 dt|scn|
+---+----+----+----+-------------------+---+
|  1|null|   1|1.12|2000-01-02 00:11:11|112|
|  1| aaa|   1|1.11|2000-01-01 00:00:00|111|
|  2| bbb|null|2.22|2000-01-03 12:12:12|201|
|  2|null|   2|1.13|               null|200|
|  2|null|null|2.33|               null|202|
|  3| ccc|   3|3.34|               null|302|
|  3|null|null|3.33|               null|301|
|  3|null|null| 0.0|2000-12-31 23:59:59|300|
+---+----+----+----+-------------------+---+

我希望获得以下DF - 按qazxsw poi排序，按qazxsw poi分组，并为每个列取最后一个非null值（scn和id除外）。

它可以这样做：

id

在现实生活中，我想处理具有大量列的不同DF。

我的问题是 - 如何以编程方式在scn中指定所有列（scala> :paste // Entering paste mode (ctrl-D to finish) df.orderBy("scn") .groupBy("id") .agg(last("name", true) as "name", last("cnt", true) as "cnt", last("amt", true) as "amt", last("dt", true) as "dt") .show // Exiting paste mode, now interpreting. +---+----+---+----+-------------------+ | id|name|cnt| amt| dt| +---+----+---+----+-------------------+ | 1| aaa| 1|1.12|2000-01-02 00:11:11| | 3| ccc| 3|3.34|2000-12-31 23:59:59| | 2| bbb| 2|2.33|2000-01-03 12:12:12| +---+----+---+----+-------------------+ 和id除外）？

生成源DF的代码：

scn

答案

解决方案类似于.agg(last(col_name, true))，并将生成的DF中的列重命名为原始列：

case class C(id: Integer, name: String, cnt: Integer, amt: Double, dt: String, scn: Integer)

val cc = Seq(
  C(1, null, 1, 1.12, "2000-01-02 00:11:11", 112),
  C(1, "aaa", 1, 1.11, "2000-01-01 00:00:00", 111),
  C(2, "bbb", null, 2.22, "2000-01-03 12:12:12", 201),
  C(2, null, 2, 1.13, null,200),
  C(2, null, null, 2.33, null, 202),
  C(3, "ccc", 3, 3.34, null, 302),
  C(3, null, null, 3.33, "20001-01-01 00:33:33", 301),
  C(3, null, null, 0.00, "2000-12-31 23:59:59", 300)
)

val t = sc.parallelize(cc, 4).toDF()
val df = t.withColumn("dt", $"dt".cast("timestamp"))
val cols = df.columns.filterNot(_.equals("id"))

结果：

this answer

要么：

val exprs = df.columns.filterNot(_.equals("id")).map(last(_, true))
val r = df.orderBy("scn").groupBy("id").agg(exprs.head, exprs.tail: _*).toDF(df.columns:_*)