PySpark merge如何实现类似pandas的indicator数据来源标识功能

pyspark.pandas的merge()方法目前没有原生支持pandas的indicator参数，直接传入会抛出错误：

TypeError: DataFrame.merge() got an unexpected keyword argument 'indicator'

如果需要在合并结果中新增列标记每行数据的来源，取值为left_only、right_only、both，可使用以下两种可行方案：

方案1：空值判断法（生产环境通用，适配所有数据规模） #

该方案和pandas原生indicator的底层实现逻辑一致，通过判断关联后左右表专属标记字段的空值情况生成来源标识，支持所有join类型（left/right/inner/outer），无内存溢出风险。
使用前先导入依赖：

from pyspark.sql import functions as F

完整实现代码（适配left join场景）：

# 给左右表分别添加不冲突的临时标记字段
left_with_flag = _left_df.withColumn("_tmp_left", F.lit(True))
right_with_flag = _right_df.withColumn("_tmp_right", F.lit(True))

# 执行原有merge逻辑
dfMerged = left_with_flag.merge(
    right_with_flag,
    how="left",
    left_on=["Col1"],
    right_on=["Col1"],
    suffixes=("_L", "_R")
)

# 根据临时标记生成来源列，最后删除临时字段
dfMerged = dfMerged.withColumn(
    "_merge",
    F.when(F.col("_tmp_left") & F.col("_tmp_right"), "both")
    .when(F.col("_tmp_left").isNotNull() & F.col("_tmp_right").isNull(), "left_only")
    .when(F.col("_tmp_left").isNull() & F.col("_tmp_right").isNotNull(), "right_only")
).drop("_tmp_left", "_tmp_right")

如果使用的是原生PySpark DataFrame而非pyspark.pandas DataFrame，仅需把代码中的merge替换为join即可，其余逻辑完全不变。

方案2：pandas中转实现（仅适合小数据集场景） #

如果待合并的数据量较小，不会超过驱动节点内存上限，可以先将pyspark.pandas DataFrame转为原生pandas DataFrame，执行带indicator参数的merge后再转回：

import pyspark.pandas as ps

# 转pandas完成带indicator的合并
pdf_merged = _left_df.to_pandas().merge(
    _right_df.to_pandas(),
    how="left",
    left_on=["Col1"],
    right_on=["Col1"],
    suffixes=("_L", "_R"),
    indicator=True
)

# 转回pyspark.pandas DataFrame
dfMerged = ps.from_pandas(pdf_merged)

注意：数据量较大时不要使用该方案，to_pandas()会把集群所有数据拉取到驱动节点，极易触发内存溢出错误。

关联类型与indicator取值对应关系 #

• how='left'：结果中仅会出现both、left_only两种取值
• how='right'：结果中仅会出现both、right_only两种取值
• how='inner'：结果中仅会出现both一种取值
• how='outer'：结果中left_only、right_only、both三种取值都可能出现

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