如何通过循环构造PySpark DataFrame的join连接条件

PySpark join 方法的连接条件参数仅接受Column类型的布尔表达式，不支持传入字符串格式的条件。

• 你通过循环拼接生成的是普通Python字符串，即便文本内容和手动编写的条件代码完全一致，也不会被解析为可执行的Column匹配逻辑，因此触发运行报错
• 注释中手动编写的条件是直接通过F.col运算生成的Column对象，符合参数类型要求，所以可以正常运行

不需要拼接代码字符串，直接在逻辑中构造Column类型的比较条件，通过&运算符累加所有关联键的等值判断即可，推荐用reduce简化循环累加逻辑：

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

key = "col1 col2 col3"

def CompareData(df1, df2, key):
    key_list = key.split(" ")
    # 为df1所有列添加x_前缀，避免和df2的同名列冲突
    df1_tmp = df1.select([F.col(c).alias(f"x_{c}") for c in df1.columns])
    
    # 累加构造Column类型的联合join条件
    key_condition = reduce(
        lambda accumulated_cond, col_name: accumulated_cond & (F.col(col_name) == F.col(f"x_{col_name}")),
        key_list,
        F.lit(True) # 初始值设为恒真条件，兼容关联键为空的边界场景
    )

    df_compare = df2.join(df1_tmp, key_condition, "left")
    # 后续可在此处补充字段值差异比对、差异标记等逻辑
    return df_compare

小提示：如果两个DataFrame做join的关联键列名完全相同，不需要加前缀区分的话，更简便的写法是直接把关联键列表传入join的on参数，Spark会自动按同名列做等值匹配。

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