PySpark group by滚动窗口

xj3cbfub 于 2023-08-06 发布在 Apache

关注(0)|答案(1)|浏览(101)

假设我有一个包含三列的表：dt、id和value。

df_tmp = spark.createDataFrame([('2023-01-01', 1001, 5),
                                ('2023-01-15', 1001, 3),
                                ('2023-02-10', 1001, 1),
                                ('2023-02-20', 1001, 2),
                                ('2023-01-02', 1002, 7),
                                ('2023-01-02', 1002, 6),
                                ('2023-01-03', 1002, 1)],
                               ["date", "id", "value"])
df.show()
# +----------+----+-----+
# |      date|  id|value|
# +----------+----+-----+
# |2023-01-01|1001|    5|
# |2023-01-15|1001|    3|
# |2023-02-10|1001|    1|
# |2023-02-20|1001|    2|
# |2023-01-02|1002|    7|
# |2023-01-02|1002|    6|
# |2023-01-03|1002|    1|
# +----------+----+-----+

字符串
我想计算value的30天滚动总和，id分组为每个date，此外，id被 * 看到 * 的一些不同日期。看起来像这样的东西：

+----------+----+-----+----------------+-------------------------+
|      date|  id|value|30_day_value_sum|days_seen_in_past_30_days|
+----------+----+-----+----------------+-------------------------+
|2023-01-01|1001|    5|               0|                        0|
|2023-01-15|1001|    3|               0|                        1|
|2023-02-10|1001|    1|               3|                        1|
|2023-02-20|1001|    2|               1|                        2|
|2023-01-02|1002|    7|               0|                        0|
|2023-01-02|1002|    6|               7|                        1|
|2023-01-03|1002|    1|              13|                        2|
+----------+----+-----+----------------+-------------------------+

型
我怀疑可以使用Window来实现，但不清楚具体细节。

apache-spark

来源：https://stackoverflow.com/questions/76831553/pyspark-group-by-with-rolling-window

1条答案

按热度按时间

bxjv4tth1#

我假设你的dataframe中最初有 date 数据类型，所以我使用了比你的输入稍微修改的输入：

from pyspark.sql import functions as F, Window as W
df_tmp = spark.createDataFrame(
    [('2023-01-01', 1001, 5),
     ('2023-01-15', 1001, 3),
     ('2023-02-10', 1001, 1),
     ('2023-02-20', 1001, 2),
     ('2023-01-01', 1002, 7),
     ('2023-01-02', 1002, 6),
     ('2023-01-03', 1002, 1)],
    ["date", "id", "value"]
).withColumn('date', F.col('date').cast('date'))

字符串
在这种情况下，以下窗口应该工作：

w = W.partitionBy('id').orderBy(F.expr("unix_date(date)")).rangeBetween(-30, -1)
df = df_tmp.withColumn('30_day_value_sum', F.sum('value').over(w)) \
           .withColumn('days_seen_in_past_30_days', F.count('id').over(w))
df = df.fillna(0, subset=['30_day_value_sum'])

df.show()
# +----------+----+-----+----------------+-------------------------+
# |      date|  id|value|30_day_value_sum|days_seen_in_past_30_days|
# +----------+----+-----+----------------+-------------------------+
# |2023-01-01|1001|    5|               0|                        0|
# |2023-01-15|1001|    3|               5|                        1|
# |2023-02-10|1001|    1|               3|                        1|
# |2023-02-20|1001|    2|               1|                        1|
# |2023-01-01|1002|    7|               0|                        0|
# |2023-01-02|1002|    6|               7|                        1|
# |2023-01-03|1002|    1|              13|                        2|
# +----------+----+-----+----------------+-------------------------+

型
如果你在“date”列中有 string 类型，那么你应该使用下面的.orderBy`子句：

.orderBy(F.expr("unix_date(to_date(date))")).rangeBetween(-30, -1)

型
更多选项和细节在这里。

赞(0）回复(0）举报 2023-08-06

我来回答

PySpark group by滚动窗口

1条答案

相关问题

热门标签

最新问答