PySpark迭代多轮Left Join时出现歧义列错误

问题描述 #

我正在编写代码，基于每轮指定的列对两个DataFrame进行多轮迭代Left Join。单轮运行正常，但第二轮基于firstname列对未匹配id的剩余行进行Join时，出现歧义列错误。

示例数据 #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType
from pyspark.sql.functions import lit, col

spark = SparkSession.builder.appName("MultiJoin").getOrCreate()

sample_data = [("Amit","","Gupta","36678","M",4000),
               ("Anita","Mathews","","40299","F",5000), 
               ("Ram","","Aggarwal","42124","M",5000),  
               ("Pooja","Anne","Goel","39298","F",5000),    
               ("Geeta","Banuwala","Brown","12345","F",-2)  
  ] 
sample_schema = StructType([
    StructField("firstname",StringType(),True),
    StructField("middlename",StringType(),True),
    StructField("lastname",StringType(),True),
    StructField("id", StringType(), True),
    StructField("gender", StringType(), True),
    StructField("salary", IntegerType(), True)
])   
df1 = spark.createDataFrame(data = sample_data, schema = sample_schema) 

sample_data = [("Amit", "ABC","MTS","36678",10),
               ("Ani", "DEF","CS","40299",200), 
               ("Ram", "ABC","MTS","421",40),   
               ("Pooja", "DEF","CS","39298",50),    
               ("Geeta", "ABC","MTS","12345",-20)   

  ] 
sample_schema = StructType([
    StructField("firstname",StringType(),True),
    StructField("Company",StringType(),True),
    StructField("position",StringType(),True),
    StructField("id", StringType(), True),
    StructField("points", IntegerType(), True)
])  
df2 = spark.createDataFrame(data = sample_data, schema = sample_schema) 

自定义Join代码（原报错版本） #

def joint_left_custom(df1, df2, cols_to_join, cols_df1_to_keep, cols_df2_to_keep):
    
    resultant_df = None
    df1_cols = df1.columns
    df2 = df2.withColumn("flag", lit(True))
    
    for i in range(len(cols_to_join)):
        joined_df = df1.join(df2, [(df1[col_1] == df2[col_2]) for col_1, col_2 in cols_to_join[i].items()], 'left')

        joined_df = joined_df.select(*[df1[column] if column in cols_df1_to_keep else df2[column] for column in cols_df1_to_keep + cols_df2_to_keep])

        df1 = (joined_df
               .filter("flag is NULL")
               .select(df1_cols)
              )
        
        resultant_df = (joined_df.filter(col("flag") == True) if i == 0 
                        else resultant_df.filter(col("flag") == True).union(resultant_df)
                       )
        
    return resultant_df

cols_to_join = [{"id": "id"}, {"firstname":"firstname"}]
cols_df1_to_keep = ["firstname", "middlename", "lastname", "id", "gender", "salary"]
cols_df2_to_keep = ["company", "position", "points"]
x = joint_left_custom(df1, df2, cols_to_join, cols_df1_to_keep, cols_df2_to_keep)

报错信息（翻译后） #

列position#29518、company#29517、points#29520存在歧义。这可能是因为你连接了多个数据集，其中某些数据集是相同的。该列指向其中一个数据集，但Spark无法确定具体是哪一个。请在连接前通过Dataset.as为数据集设置不同的别名，并使用限定名称指定列，例如df.as("a").join(df.as("b"), $"a.id" > $"b.id")。你也可以将spark.sql.analyzer.failAmbiguousSelfJoin设置为false来禁用此检查。

问题分析 #

1. 列名冲突与歧义：第二轮循环时，df1已经是第一轮Join后的结果（包含df2的列），再次和原始df2Join时，firstname等列同时存在于两个DataFrame中，Spark无法区分来源。
2. select逻辑不严谨：原代码中select判断列是否在cols_df1_to_keep时，未考虑列名大小写问题（比如df2中是Company，但cols_df2_to_keep写的是company），且未用别名明确指定列来源。
3. resultant_df的union操作隐患：多次union时，若列的来源未统一，会加剧歧义问题。

解决方案 #

1. 为DataFrame设置别名：每次Join前给df1和df2分别设置别名，通过别名明确列的来源。
2. 统一列名大小写：将df2的Company列重命名为company，与cols_df2_to_keep中的名称一致。
3. 明确指定列来源：在join条件、select、filter操作中，全部使用别名引用列。
4. 清理剩余行的列：确保每次循环后，剩余未匹配的df1只保留原始df1的列，避免携带上一轮的df2列。

修正后的代码 #

def joint_left_custom(df1, df2, cols_to_join, cols_df1_to_keep, cols_df2_to_keep):
    resultant_df = None
    df1_cols = df1.columns
    # 统一df2的列名大小写，避免匹配失败
    df2 = df2.withColumnRenamed("Company", "company").withColumn("flag", lit(True)).alias("right")
    
    for i in range(len(cols_to_join)):
        # 给当前df1设置别名
        df1_alias = df1.alias("left")
        # 构建带别名的join条件
        join_conditions = [col(f"left.{col1}") == col(f"right.{col2}") for col1, col2 in cols_to_join[i].items()]
        joined_df = df1_alias.join(df2, join_conditions, 'left')
        
        # 明确指定列的来源，避免歧义
        select_cols = []
        for col_name in cols_df1_to_keep:
            select_cols.append(col(f"left.{col_name}").alias(col_name))
        for col_name in cols_df2_to_keep:
            select_cols.append(col(f"right.{col_name}").alias(col_name))
        # 保留flag列用于筛选匹配行
        select_cols.append(col("right.flag"))
        joined_df = joined_df.select(*select_cols)
        
        # 更新df1为未匹配的剩余行，只保留原始df1的列
        df1 = joined_df.filter(col("flag").isNull()).select(df1_cols)
        
        # 合并匹配结果
        matched_df = joined_df.filter(col("flag") == True).drop("flag")
        if i == 0:
            resultant_df = matched_df
        else:
            resultant_df = resultant_df.union(matched_df)
    
    return resultant_df

cols_to_join = [{"id": "id"}, {"firstname":"firstname"}]
cols_df1_to_keep = ["firstname", "middlename", "lastname", "id", "gender", "salary"]
cols_df2_to_keep = ["company", "position", "points"]
x = joint_left_custom(df1, df2, cols_to_join, cols_df1_to_keep, cols_df2_to_keep)
x.show()

验证结果 #

修正后的代码运行后，会先通过id匹配到Amit、Pooja、Geeta，再通过firstname匹配到Ram，最终输出所有匹配成功的行，无歧义列错误。

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