假设我们有以下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-由scn
排序,由id
抓取,并为每列的最后一个不null值(id
和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|
+---+----+---+----+-------------------+
在现实生活中,我想用大量列处理不同的DF。
我的问题是 - 如何在 生成源DF的代码:.agg(last(col_name, true))
中指定所有列(id
和scn
除外)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"))
解决方案类似于此答案,以及最初的DF中的重命名列:
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:_*)
结果:
scala> r.show
+---+----+---+----+-------------------+---+
| id|name|cnt| amt| dt|scn|
+---+----+---+----+-------------------+---+
| 1| aaa| 1|1.12|2000-01-02 00:11:11|112|
| 3| ccc| 3|3.34|2000-12-31 23:59:59|302|
| 2| bbb| 2|2.33|2000-01-03 12:12:12|202|
+---+----+---+----+-------------------+---+
或:
val exprs = df.columns.filterNot(_.equals("id")).map(c=>last(c, true).as(c.toString))
val r = df.orderBy("scn").groupBy("id").agg(exprs.head, exprs.tail: _*)