PySpark如何合并配置与用户输入的动态过滤条件实现单步执行

完全可以把两步过滤合并为一步，也支持将动态参数写入配置统一维护，两种常用实现方案如下：

方案1：运行时拼接两类过滤条件 #

不需要修改原有配置结构，直接在运行时把配置规则和动态日期规则拼接后一次过滤：

• 第一步：提取配置中的固定过滤规则
• 第二步：构造动态日期区间的过滤表达式
• 第三步：拼接两类规则后调用一次filter执行过滤

可运行代码示例 #

from pyspark.sql.functions import date_sub
import json

# 测试数据
df = spark.createDataFrame(
    [
        (123,"2020-01-01"),
        (124,"2020-01-01"),
        (123,"2019-01-01")
    ],
    ("cust", "activity_day")
)
# 解析配置
config = json.loads("""
[ {
    "source":"df_1",
    "filters":"cust=123"
}
]
""")
config_filter = config[0]['filters']

# 用户传入参数
rundate = "2020-01-02"
history_days = 2

# 构造合并后的过滤条件
date_filter = f"activity_day >= date_sub('{rundate}', {history_days}) AND activity_day < '{rundate}'"
final_filter = f"{config_filter} AND {date_filter}"

# 一次过滤完成
result_df = df.filter(final_filter)
result_df.show()

运行后输出符合要求的结果：

+----+-------------+
|cust|activity_day|
+----+-------------+
| 123|   2020-01-01|
+----+-------------+

方案2：配置中预留占位符统一维护所有规则 #

如果希望所有过滤规则都在配置文件中统一管理，可以在配置的过滤规则里预留动态参数占位符，运行时替换后直接使用：
修改配置示例：

[ {
    "source":"df_1",
    "filters":"cust=123 AND activity_day >= date_sub('${rundate}', ${history_days}) AND activity_day < '${rundate}'"
}
]

运行时代码：

# 提取配置规则后替换占位符
raw_filter = config[0]['filters']
final_filter = raw_filter.replace("${rundate}", rundate).replace("${history_days}", str(history_days))

# 直接执行过滤
result_df = df.filter(final_filter)

注意事项 #

• 如果配置中存在多条过滤规则，统一用AND拼接后再和日期规则合并即可
• 若输入参数来自不可信的外部来源，建议用PySpark的参数化表达式代替直接字符串拼接，避免SQL注入风险
• 可以提前把activity_day字段转成日期类型，避免字符串比对出现异常

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