假设我有 5 TB 的数据具有以下模式,并且我正在使用 Pyspark。
| id | date | Month | KPI_1 | ... | KPI_n
对于 90% 的 KPI,我只需要知道总和/最小/最大值聚合到(id、月(级别。对于其余的 10%,我需要知道基于日期的第一个值。
我的一种选择是使用window
.例如,我可以做
from pyspark.sql import Window
import pyspark.sql.functions as F
w = Window.partitionBy("id", "Month").orderBy(F.desc("date"))
# for the 90% kpi
agg_df = df.withColumn("kpi_1", F.sum("kpi_1").over(w))
agg_df = agg_df.withColumn("kpi_2", F.max("kpi_2").over(w))
agg_df = agg_df.withColumn("kpi_3", F.min("kpi_3").over(w))
...
# Select last row for each window to get last accumulated sum for 90% kpis and last value for 10% kpi (which is equivalent to first value if ranked ascending).
# continue process agg_df with filters based on sum/max/min values of 90% KIPs.
但是我不确定如何选择每个窗口的最后一行。有没有人有任何建议,或者是否有更好的聚合方法?
假设我们有这些数据
+---+----------+-------+-----+-----+
| id| date| month|kpi_1|kpi_2|
+---+----------+-------+-----+-----+
| 1|2000-01-01|2000-01| 1| 100|
| 1|2000-01-02|2000-01| 2| 200|
| 1|2000-01-03|2000-01| 3| 300|
| 1|2000-01-04|2000-01| 4| 400|
| 1|2000-01-05|2000-01| 5| 500|
| 1|2000-02-01|2000-02| 10| 11|
| 1|2000-02-02|2000-02| 20| 21|
| 1|2000-02-03|2000-02| 30| 31|
| 1|2000-02-04|2000-02| 40| 41|
+---+----------+-------+-----+-----+
我们想计算kpi_1
的最小值、最大值和总和,并得到每个组的最后一个值kpi_2
。
获取kpi_1
的最小值、最大值和总和可以通过按id
和month
对数据进行分组来实现。使用Spark>= 3.0.0 max_by可用于获取kpi_2
的最新值:
df_avg = df
.groupBy("id","month")
.agg(F.sum("kpi_1"), F.min("kpi_1"), F.max("kpi_1"), F.expr("max_by(kpi_2, date)"))
df_avg.show()
指纹
+---+-------+----------+----------+----------+-------------------+
| id| month|sum(kpi_1)|min(kpi_1)|max(kpi_1)|max_by(kpi_2, date)|
+---+-------+----------+----------+----------+-------------------+
| 1|2000-02| 100| 10| 40| 41|
| 1|2000-01| 15| 1| 5| 500|
+---+-------+----------+----------+----------+-------------------+
对于Spark 版本 <3.0.0max_by
不可用,因此获取每个组的最后一个值kpi_2
更加困难。第一个想法可能是在降序有序数据帧上使用聚合函数 first((。一个简单的测试给了我正确的结果,但不幸的是,文档指出"该函数是不确定的,因为它的结果取决于行的顺序,这些行在洗牌后可能是不确定的"。
获取kpi_2
最后一个值的更好方法是使用问题中显示的窗口。作为窗口函数 row_number(( 将工作:
w = Window.partitionBy("id", "Month").orderBy(F.desc("date"))
df_first = df.withColumn("row_number", F.row_number().over(w)).where("row_number = 1")
.drop("row_number")
.select("id", "month", "KPI_2")
df_first.show()
指纹
+---+-------+-----+
| id| month|KPI_2|
+---+-------+-----+
| 1|2000-02| 41|
| 1|2000-01| 500|
+---+-------+-----+
连接第一部分(不带max_by
列(和第二部分将得到所需的结果:
df_result = df_avg.join(df_first, ['id', 'month'])
df_result.show()
指纹
+---+-------+----------+----------+----------+-----+
| id| month|sum(kpi_1)|min(kpi_1)|max(kpi_1)|KPI_2|
+---+-------+----------+----------+----------+-----+
| 1|2000-02| 100| 10| 40| 41|
| 1|2000-01| 15| 1| 5| 500|
+---+-------+----------+----------+----------+-----+