Pyspark Dataframe - 每个月的最后一天计算 baklog答案

【问题标题】：Pyspark Dataframe - Counting baklog each last day of the monthPyspark Dataframe - 每个月的最后一天计算 baklog
【发布时间】：2021-09-22 11:26:36
【问题描述】：

我正在尝试寻找解决一个问题的方法。我有以下df：

task	open_date	close_date
t01	05/03/2021	16/03/2021
t02	07/03/2021	13/04/2021
t03	23/03/2021	04/04/2021

我需要一个新的 df，月底有积压（当月打开，之后关闭）：

end_of_month	backlog
31/03/2021	2

我正在使用 last_day() 来获取 EOM，但我没有使用查询/代码来获取正确的计数。你可以帮帮我吗？提前致谢！！

【问题讨论】：

标签： python dataframe pyspark count backlog

【解决方案1】：

    df
    .withColumn('open_month_next_month', add_months('open_date', 1))
    .filter(last_day('open_month_next_month') == last_day('close_date'))
    .withColumn('end_of_month', last_day('open_date'))
    .groupBy('end_of_month')
    .agg(F.count(F.lit(1)).alias('backlog'))

【讨论】：

嗨 NaWeeD，感谢您的回答，但我想这仅适用于截止日期是下个月但可能在 2 个月或更长时间之后。
@David 好吧，我想您应该只将 == 更改为 <= 对吧？

【解决方案2】：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

data = [("t01", "05/03/2021", "16/03/2021"),
        ("t02", "07/03/2021", "13/04/2021"),
        ("t03", "23/03/2021", "04/04/2021"),
        ("t04", "07/03/2021", "13/05/2021"),
        ("t05", "23/03/2021", "04/05/2021")]
schema = ["TaskID", "TaskstartDate", "TaskEndDate"]

df = spark.createDataFrame(data, schema)

df = df.withColumn("TaskstartDate",F.to_date(F.col("TaskstartDate"), "dd/MM/yyyy"))\
.withColumn("TaskEndDate",F.to_date(F.col("TaskEndDate"), "dd/MM/yyyy"))

df_grouped = df.withColumn("EndDay_fromStartDate", F.last_day(F.col("TaskstartDate")))\
.withColumn("EndDay_fromEndDate", F.last_day(F.col("TaskEndDate")))\
.filter((F.col("EndDay_fromStartDate") != F.col("EndDay_fromEndDate")) & (F.col("TaskstartDate") < F.col("TaskEndDate")))\
.groupBy('EndDay_fromStartDate')\
.agg(F.count(F.lit(1)).alias('backlog'))

df_grouped.show()

【讨论】：

谢谢克里斯！我认为这正是我所需要的。我会尝试一下，如果它工作正常，我会将其标记为解决方案。
最后，它根本不起作用。当我有一个在 3 月打开并在 6 月关闭的任务时，它只会出现在 3 月的积压中，而且必须在 4 月和 5 月出现。