基于多DataFrame与指定日期范围设置列值的实现方案

问题背景 #

• 现有两个数据集：
  • RENTAL_DF：记录租赁单元每日状态，包含DATE_KEY、PRODUCT_KEY、FISCAL_YR、RENTAL_NM、OCCUPIED_IND五列，时间范围覆盖20190901至20221221
  • ACTIVE_AUDIT_DF：记录租赁单元的激活状态变更时间，但无法直接与RENTAL_DF的日期做精准匹配
• 核心需求：为RENTAL_DF新增ACTIVE_IND列，标记每个租赁单元每日的激活状态，最终生成FINAL_DF

解决方案（以PySpark为例） #

1. 预处理状态变更审计数据 #

首先把ACTIVE_AUDIT_DF的单条变更记录转化为状态生效的时间区间，假设该数据集包含PRODUCT_KEY（租赁单元ID）、CHANGE_DATE（变更生效日期）、NEW_ACTIVE_STATUS（变更后的激活状态，比如Y/N）：

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

# 按租赁单元分组，按变更日期排序，获取下一次状态变更的日期
audit_window = Window.partitionBy("PRODUCT_KEY").orderBy("CHANGE_DATE")
processed_audit = ACTIVE_AUDIT_DF.withColumn(
    "NEXT_CHANGE_DATE",
    F.lead("CHANGE_DATE").over(audit_window)
)

# 处理最后一次变更的结束日期（设为RENTAL_DF的最大日期）
max_rental_date = RENTAL_DF.select(F.max("DATE_KEY")).first()[0]
processed_audit = processed_audit.withColumn(
    "NEXT_CHANGE_DATE",
    F.coalesce("NEXT_CHANGE_DATE", F.lit(max_rental_date))
)

2. 关联每日状态数据与激活区间 #

通过租赁单元ID+日期区间的方式，将激活状态映射到RENTAL_DF的每条每日记录：

final_df = RENTAL_DF.join(
    processed_audit,
    (RENTAL_DF.PRODUCT_KEY == processed_audit.PRODUCT_KEY) &
    (RENTAL_DF.DATE_KEY >= processed_audit.CHANGE_DATE) &
    (RENTAL_DF.DATE_KEY < processed_audit.NEXT_CHANGE_DATE),
    how="left"
).select(
    RENTAL_DF["*"],
    # 无匹配变更记录时设为默认状态，这里默认未激活，可根据业务调整
    F.coalesce("NEW_ACTIVE_STATUS", F.lit("N")).alias("ACTIVE_IND")
)

3. 业务规则校验 #

因为租赁单元仅激活时可被占用，可添加校验确保数据符合业务逻辑：

# 检查未激活状态下是否存在被占用的异常记录
final_df.filter(
    (final_df.ACTIVE_IND == "N") & (final_df.OCCUPIED_IND == "Y")
).show()

关键注意事项 #

• 先统一两个数据集的日期格式：确保DATE_KEY和CHANGE_DATE格式一致（比如都是yyyyMMdd整数或标准日期类型）
• 默认状态需贴合业务：如果租赁单元初始为激活状态，可把默认值改为Y
• Pandas环境下可替代方案：使用pd.merge_asof（需排序）或区间索引连接实现相同逻辑

内容的提问来源于stack exchange，提问作者David Kurtenbach