PySpark如何实现基于参考DataFrame的动态like条件过滤

问题场景 #

基于参考DataFrame中同ID对应的Value值，对目标DataFrame做包含匹配，硬编码OR条件的方式无法适配匹配值动态增减的场景，以下两种实现都可以做到零硬编码自动适配数据变化。

注：你提供的硬编码示例存在笔误，后两个LIKE条件错误指定了ID列，实际需要匹配Value列，以下实现均已修正该问题，匹配逻辑符合业务预期。

方案1：动态生成OR条件（适配小数据量参考集） #

实现逻辑：

• 先按ID分组，收集每个ID下所有需要匹配的Value值
• 遍历收集到的匹配规则，自动拼接like('%xxx%')的OR逻辑
• 合并所有ID的过滤规则完成筛选

代码实现：

from pyspark.sql import functions as F
from functools import reduce

# 按ID分组收集所有待匹配值
id_value_map = df.groupBy("ID")\
    .agg(F.collect_list("Value").alias("match_vals"))\
    .rdd.collectAsMap()

# 动态生成每个ID的过滤条件
condition_list = []
for target_id, vals in id_value_map.items():
    # 拼接当前ID下所有Value的LIKE OR条件
    value_match = reduce(
        lambda x, y: x | y,
        [F.col("Value").like(f"%{v}%") for v in vals]
    )
    condition_list.append( (F.col("ID") == target_id) & value_match )

# 合并所有ID的条件执行过滤
final_condition = reduce(lambda x, y: x | y, condition_list)
res_df = final_df.filter(final_condition)
res_df.show(5, False)

运行输出：

+---+----------+
|ID |Value     |
+---+----------+
|1  |1234563478|
|2  |2134510   |
|3  |789033323 |
+---+----------+

方案2：LEFT SEMI JOIN实现（适配大数据量参考集） #

如果参考DataFrame的匹配值总量过万，拼接超长OR条件会导致Spark SQL解析效率极低，这时候用半连接实现性能更好，逻辑等价于过滤：

from pyspark.sql import functions as F

# 构造匹配规则表，提前生成LIKE匹配串
match_rule_df = df.select(
    F.col("ID").alias("m_id"),
    F.concat(F.lit("%"), F.col("Value"), F.lit("%")).alias("pattern")
)

# 左半连接实现过滤，不会产生重复数据
res_df = final_df.join(
    match_rule_df,
    on=(
        (F.col("ID") == F.col("m_id")) &
        (F.col("Value").like(F.col("pattern")))
    ),
    how="left_semi"
)

res_df.show(5, False)

运行结果和方案1完全一致。

方案选型建议 #

• 参考集匹配值总量在千级以内：选方案1，逻辑直观易调试
• 参考集匹配值过万：选方案2，执行效率更高，无SQL解析瓶颈

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