Pyspark - 添加缺少日期的行并用 0 填充值答案

【问题标题】：Pyspark - Add rows for missing dates and fill value with 0Pyspark - 添加缺少日期的行并用 0 填充值
【发布时间】：2020-06-10 04:30:50
【问题描述】：

我目前正在跟踪我的产品中的用户每月计数。这个问题是当用户在特定月份没有任何活动时，我将丢失行。这是一个例子：

最小月份：

+---------------+
|min(year_month)|
+---------------+
|        2019_05|
+---------------+

最大月份：

+---------------+
|max(year_month)|
+---------------+
|        2020_06|
+---------------+

用户数据：

+--------------------+----------+----------------------+
|             core_id|year_month|month_sum_detailaction|
+--------------------+----------+----------------------+
|000006c9-d42b-4fe...|   2019_09|                     3|
|000006c9-d42b-4fe...|   2020_01|                     2|
|000006c9-d42b-4fe...|   2020_02|                     6|
+--------------------+----------+----------------------+

如您所见，该用户在 12 个月中只有 3 个月有活动。

我想做的是将每个用户的数据更新为如下所示：

+--------------------+----------+----------------------+
|             core_id|year_month|month_sum_detailaction|
+--------------------+----------+----------------------+
|000006c9-d42b-4fe...|   2019_05|                     0|
|000006c9-d42b-4fe...|   2020_06|                     0|
|000006c9-d42b-4fe...|   2020_07|                     0|
|000006c9-d42b-4fe...|   2020_08|                     0|
|000006c9-d42b-4fe...|   2019_09|                     3|
|000006c9-d42b-4fe...|   2020_10|                     0|
|000006c9-d42b-4fe...|   2020_11|                     0|
|000006c9-d42b-4fe...|   2019_12|                     0|
|000006c9-d42b-4fe...|   2020_01|                     2|
|000006c9-d42b-4fe...|   2020_02|                     6|
|000006c9-d42b-4fe...|   2020_03|                     0|
|000006c9-d42b-4fe...|   2020_04|                     0|
|000006c9-d42b-4fe...|   2020_05|                     0|
|000006c9-d42b-4fe...|   2020_06|                     0|
+--------------------+----------+----------------------+

我对 pyspark 比较陌生，因此非常感谢任何帮助。

【问题讨论】：

这样的事情可以让你开始：stackoverflow.com/questions/39271374/…
这能回答你的问题吗？ Create PySpark dataframe : sequence of months with year

标签： dataframe pyspark

【解决方案1】：

试试这个。 你也可以collect() max 和 min year_month 放到sequence函数中。

from pyspark.sql import functions as F

df.groupBy("core_id").agg(F.collect_list(F.to_date("year_month","yyyy_MM")).alias("year_month"),\
                          F.collect_list("month_sum_detailaction").alias("month_sum_detailaction"))\
  .withColumn("seq", F.expr("""sequence(to_date('2019_05','yyyy_MM'),to_date('2020_06','yyyy_MM'),interval 1 month)"""))\
  .withColumn("year_month", F.flatten(F.array("year_month", F.array_except("seq","year_month"))))\
  .withColumn("zip", F.explode(F.arrays_zip("year_month"\
                                 ,"month_sum_detailaction")))\
  .select("core_id", F.col("zip.*"))\
  .withColumn("year_month", F.date_format("year_month", "yyyy_MM")).orderBy("year_month").fillna(0).show()

#+--------------------+----------+----------------------+
#|             core_id|year_month|month_sum_detailaction|
#+--------------------+----------+----------------------+
#|000006c9-d42b-4fe...|   2019_05|                     0|
#|000006c9-d42b-4fe...|   2019_06|                     0|
#|000006c9-d42b-4fe...|   2019_07|                     0|
#|000006c9-d42b-4fe...|   2019_08|                     0|
#|000006c9-d42b-4fe...|   2019_09|                     3|
#|000006c9-d42b-4fe...|   2019_10|                     0|
#|000006c9-d42b-4fe...|   2019_11|                     0|
#|000006c9-d42b-4fe...|   2019_12|                     0|
#|000006c9-d42b-4fe...|   2020_01|                     2|
#|000006c9-d42b-4fe...|   2020_02|                     6|
#|000006c9-d42b-4fe...|   2020_03|                     0|
#|000006c9-d42b-4fe...|   2020_04|                     0|
#|000006c9-d42b-4fe...|   2020_05|                     0|
#|000006c9-d42b-4fe...|   2020_06|                     0|
#+--------------------+----------+----------------------+

【讨论】：