另一种根据日期值获取最新记录的方法

Question

我有一个输入日期，但示例如下：df_inp

customer_id  |ph_num|date      |
1            |123   |2020-10-01|
2            |456   |2020-10-01|
3            |789   |2020-10-01|
1            |654   |2020-10-02|
2            |543   |2020-10-03|
1            |908   |2020-10-04|
4            |123   |2020-10-02|

我需要获取每个日常流程的最新记录。所以，我尝试了 windows rank() 操作并且它有效。但是，随着输入数据以百万计来优化性能，我们可以使用任何其他 spark 操作来获取基于 customer_id 和日期值排序的最新数据。

window_func = Window.partition_by("customer_id ").orderBy("date")
df = df.withColumn("rank", rank().over(window_func))
df = df.filter(df.rank == "1")

这里是 customer_id - 字符串和日期 - 时间戳

Answer 1

对于 Spark 3.0+，可能值得检查 max_by（或 min_by，如果您在问题中排名 1）比 window + filter 方法具有更好的性能特征。

df.groupBy("customer_id").agg(F.expr("max_by(ph_num,date)"), F.max(F.col("date")))

结果与问题中的相同。比较两种方法的执行计划，max_by 方法少了一个转换（filter），但两种方法都会触发一次交换。