【发布时间】:2021-11-16 22:47:03
【问题描述】:
假设我有一个如下所示的数据集:
val data1 = Seq(
("81518165", "10", "0412221432", "2021.02.01 12:29:57"),
("81518165", "10", "0412392873", "2021.02.01 11:33:41"),
("81518165", "10", "0412392879", "2021.02.01 05:12:12"),
("81518165", "10", "0412392950", "2021.02.01 01:39:37"),
("23698102", "12", "0412221432", "2021.02.14 12:55:33"),
("23698102", "12", "0412392873", "2021.02.14 11:33:37"),
("23698102", "12", "0412392879", "2021.02.14 05:12:00")
)
val df1 = data1.toDF("AUFTRAG", "AUFTRAG_POS", "IID_CODE", "ERST_TIMESTAMP")
我想删除由日期引起的重复行,方法是根据最大日期“ERST_TIMESTAMP”聚合“AUFTRAG”和“AUFTRAG_POS”列。 要获得最大日期,这是我的代码:
df1.withColumn("ERST_TIMESTAMP", to_timestamp(col("ERST_TIMESTAMP"),"yyyy.MM.dd HH:mm:ss"))
.groupBy("AUFTRAG", "AUFTRAG_POS")
.agg(max("ERST_TIMESTAMP"))
.show()
这是预期的结果:
+--------+-----------+-------------------+
| AUFTRAG|AUFTRAG_POS|max(ERST_TIMESTAMP)|
+--------+-----------+-------------------+
|81518165| 10|2021-02-01 12:29:57|
|23698102| 12|2021-02-14 12:55:33|
+--------+-----------+-------------------+
我现在的目标是在这个最大日期之前替换按“AUFTRAG”和“AUFTRAG_POS”分组的 ERST_TIMESTAMP。这是我的解决方案:
val df2 = df1.withColumn("ERST_TIMESTAMP", to_timestamp(col("ERST_TIMESTAMP"),"yyyy.MM.dd HH:mm:ss"))
.groupBy("AUFTRAG", "AUFTRAG_POS")
.agg(max("ERST_TIMESTAMP"))
df1.join(df2, Seq("AUFTRAG", "AUFTRAG_POS")).show()
预期的结果,正是我想要的:
我对这种方法不是很满意。还有其他方法吗?有什么帮助吗?
【问题讨论】:
标签: scala dataframe apache-spark group-by aggregate