如何在PySpark中动态传递多列名实现动态Left Anti Join？

一、原函数的优化点 #

原函数存在几处需要修正的问题，同时可以优化兼容性：

1. 可变默认参数陷阱：Python中cond=[]这类可变默认参数会在函数定义时创建一次，后续调用会复用同一个列表，导致意外的状态保留，需改为cond=None并在函数内初始化。
2. 未明确的依赖项：curate_path和logger如果不是全局变量，应作为参数传入函数，保证函数的独立性和可复用性。
3. 边界条件缺失：当cond为空时，df.select(cond)会触发错误，需要增加校验逻辑。
4. 拼写修正：原函数名integraty_check应为integrity_check（拼写错误）。

优化后的函数代码：

def integrity_check(testdata, refdata, cond=None, curate_path=None, logger=None):
    # 处理默认参数，规避可变默认参数陷阱
    cond = cond or []
    # 校验连接条件合法性
    if not cond:
        raise ValueError("连接条件cond不能为空，请传入至少一个列名")
    # 执行leftanti join操作
    df = func.join_dataframe(testdata, refdata, cond, "leftanti", logger)
    # 筛选指定连接列
    df = df.select(cond)
    # 写入Parquet文件（确保路径已传入）
    if curate_path:
        func.write_df_as_parquet_file(df, curate_path, logger)
    return df

二、动态传递列名的调用方式 #

调用时直接传入列名列表即可，支持单列或多列场景：

1. 单列连接条件：

# 传入单列名组成的列表
result = integrity_check(test_df, ref_df, cond=["user_id"], curate_path="./output", logger=my_logger)

2. 多列连接条件：

# 传入多列名组成的列表
result = integrity_check(test_df, ref_df, cond=["user_id", "order_date"], curate_path="./output", logger=my_logger)

可选：兼容单个字符串输入 #

如果担心调用者误传单个字符串而非列表，可以在函数内增加兼容逻辑，自动将单个字符串转为列表：

def integrity_check(testdata, refdata, cond=None, curate_path=None, logger=None):
    # 处理默认参数与单个字符串输入
    if cond is None:
        cond = []
    elif isinstance(cond, str):
        cond = [cond]
    if not cond:
        raise ValueError("连接条件cond不能为空，请传入至少一个列名")
    # 后续逻辑不变
    df = func.join_dataframe(testdata, refdata, cond, "leftanti", logger)
    df = df.select(cond)
    if curate_path:
        func.write_df_as_parquet_file(df, curate_path, logger)
    return df

此时可直接传入单个列名字符串：

result = integrity_check(test_df, ref_df, cond="user_id", curate_path="./output", logger=my_logger)

三、最优方案总结 #

• 修复可变默认参数陷阱，保证函数多次调用的一致性。
• 将外部依赖转为参数，提升函数的可维护性和复用性。
• 增加边界校验，避免空连接条件导致的运行错误。
• 可选支持单个字符串输入，提升调用灵活性。

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