sql - PySpark 数据框的每日预测

Question

我在 PySpark 中有以下数据框:

DT_BORD_REF : 月份的日期列
REF_DATE : 分隔过去和 future 的当前日期的日期引用
PROD_ID : 产品编号
COMPANY_CODE : 公司编号
CUSTOMER_CODE : 客户编号
MTD_WD :本月至今的工作日计数(日期 = DT_BORD_REF)
QUANTITY : 售出数量
QTE_MTD : 本月至今的项目数

+-------------------+-------------------+-----------------+------------+-------------+-------------+------+--------+-------+
|        DT_BORD_REF|           REF_DATE|          PROD_ID|COMPANY_CODE|CUSTOMER_CODE|COUNTRY_ALPHA|MTD_WD|QUANTITY|QTE_MTD|
+-------------------+-------------------+-----------------+------------+-------------+-------------+------+--------+-------+
|2020-11-02 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|     1|     4.0|    4.0|
|2020-11-05 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|     3|    null|    4.0|
|2020-11-06 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|     4|    null|    4.0|
|2020-11-09 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|     5|    null|    4.0|
|2020-11-10 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|     6|    null|    4.0|
|2020-11-11 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|     7|    null|    4.0|
|2020-11-12 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|     8|    null|    4.0|
|2020-11-13 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|     9|    null|    4.0|
|2020-11-16 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|    10|    null|    4.0|
|2020-11-17 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|    11|    null|    4.0|
|2020-11-18 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|    12|    null|    4.0|
|2020-11-19 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|    13|    null|    4.0|
|2020-11-20 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|    14|    null|    4.0|
|2020-11-23 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|    15|    null|    4.0|
|2020-11-24 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|    16|    null|    4.0|
|2020-11-25 00:00:00|2020-11-04 00:00:00|          0000043|         503|     KDAI3982|          RUS|    17|    null|    4.0|

DT_BORD_REF < REF_DATE所有行都是实际销售额，不一定每个工作日都会发生。有时也在非工作日发生。

DT_BORD_REF >= REF_DATE没有销售(这是 future )

目标是使用以下公式预测所有 future 行的销售额:QTE_MTD/MTD_WD在 REF_DATE 上计算针对每个产品、客户和国家/地区。

QTE_MTD 是使用窗口函数从 QUANTITY 列计算得出的。我需要将其划分为 MTD_WD在 REF_DATE在这个例子中是 3

如何使用 MTD_WD 添加一列在 REF_DATE按产品、客户和国家划分？

换句话说，我需要添加一个第一次出现 MTD_WD 的列当条件 DT_BORD_REF > REF_DATE每个产品、客户和国家/地区都满足(同样，在本示例中为 3)。

此数据集包含数百万行，分别针对不同的产品、客户和国家/地区 工作日按国家规定

希望它是清楚的:)

Answer 1

您可以将 first 与 ignorenulls=True 一起使用，并将 when 与适当的条件一起使用，以获得第一个 MTD_WD 其中 DT_BORD_REF > REF_DATE:

from pyspark.sql import functions as F, Window

df2 = df.withColumn(
    'val',
    F.first(
        F.when(
            F.col('DT_BORD_REF') > F.col('REF_DATE'),
            F.col('MTD_WD')
        ), 
        ignorenulls=True
    ).over(
        Window.partitionBy('PROD_ID','COMPANY_CODE','CUSTOMER_CODE','COUNTRY_ALPHA')
              .orderBy('DT_BORD_REF')
              .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)
    )
)