同列DataFrame对比添加last_change_date列遇字段解析错误求助

需求 #

对比两个列结构一致的DataFrame，为df1添加last_change_date相关列：当df1与df2的非关联键字段不一致时，该列取值为当前时间戳；否则沿用df2的last_change_date字段值。

示例数据 #

data_df = [("John", 25, "Male", "Engineer", 1),
           ("Alice", 30, "Female", "Doctor", 2)]

data_df2 = [("1","John", 25, "Male", "Engineer", "2023-01-01"),
            ("2","Alice", 30, "Female", "Doctor", "2023-01-02")]

尝试代码 #

def compare_dataframes(df1, df2, key_fields):
    df1_aliases = [col(f"{field}").alias(f"{field}_df1") for field in df1.columns if field not in key_fields]
    df2_aliases = [col(f"{field}").alias(f"{field}_df2") for field in df2.columns if field not in key_fields]
    join_conditions = [col(f"{field}_df1") == col(f"{field}_df2") for field in key_fields]
    result_df = df1.select(*df1.columns, *df1_aliases).alias("df1").join(
        df2.select(*df2.columns, *df2_aliases).alias("df2"),
        on=join_conditions,
        how="inner"
    )
    for field in df1.columns:
        if field not in key_fields:
            result_df = result_df.withColumn(
                f"last_change_date_{field}",
                when(
                    (col(f"{field}_df1") != col(f"{field}_df2")) | col(f"last_change_date_{field}_df1").isNull(),
                    current_timestamp()
                ).otherwise(col(f"last_change_date_{field}_df2"))
            )

    return result_df

报错信息 #

[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column or function parameter with name `Key_df1` cannot be resolved. Did you mean one of the following? [`df1`.`Age`, `df2`.`Age`, `df1`.`Age_df1`, `df1`.`Key`, `df2`.`Key`].;
'Join Inner, ('Key_df1 = 'Key_df2)
:- SubqueryAlias df1
:  +- Project [Name#6424, Age#6425L, Key#6428L, Name#6424 AS Name_df1#6483, Age#6425L AS Age_df1#6484L]
:     +- Project [Name#6424, Age#6425L, Key#6428L]
:        +- LogicalRDD [Name#6424, Age#6425L, Gender#6426, Occupation#6427, Key#6428L], false
+- SubqueryAlias df2
   +- Project [Key#6434, Name#6435, Age#6436L, last_change_date#6439, Name#6435 AS Name_df2#6485, Age#6436L AS Age_df2#6486L, last_change_date#6439 AS last_change_date_df2#6487]
      +- Project [Key#6434, Name#6435, Age#6436L, last_change_date#6439]
         +- LogicalRDD [Key#6434, Name#6435, Age#6436L, Gender#6437, Occupation#6438, last_change_date#6439], false.

问题 #

循环处理字段时出现字段解析错误，如何修改代码实现需求？

解决方案 #

错误原因 #

1. 关联条件错误：代码错误地对关联键字段生成了别名，并使用别名作为关联条件，但关联键并未被添加别名，导致字段无法解析。
2. 无效字段引用：代码中引用了last_change_date_{field}_df1，但df1本身不存在该字段，逻辑完全错误。
3. 冗余列选择：select(*df1.columns, *df1_aliases)重复选择原列和别名列，造成冗余。

修改后代码 #

from pyspark.sql import functions as F

def compare_dataframes(df1, df2, key_fields):
    # 为df1的非关联键字段添加别名，避免join后列名冲突
    df1_alias = df1.select(
        *[F.col(field) for field in key_fields],
        *[F.col(field).alias(f"{field}_df1") for field in df1.columns if field not in key_fields]
    ).alias("df1")
    
    # 为df2的非关联键字段添加别名，保留last_change_date字段
    df2_alias = df2.select(
        *[F.col(field) for field in key_fields],
        *[F.col(field).alias(f"{field}_df2") for field in df2.columns if field not in key_fields + ["last_change_date"]],
        F.col("last_change_date")
    ).alias("df2")
    
    # 使用原关联键字段构建关联条件
    join_conditions = [F.col(f"df1.{field}") == F.col(f"df2.{field}") for field in key_fields]
    
    # 执行内关联
    result_df = df1_alias.join(df2_alias, on=join_conditions, how="inner")
    
    # 为每个非关联键字段生成对应的last_change_date列
    for field in df1.columns:
        if field not in key_fields:
            result_df = result_df.withColumn(
                f"last_change_date_{field}",
                F.when(
                    F.col(f"{field}_df1") != F.col(f"{field}_df2"),
                    F.current_timestamp()
                ).otherwise(F.col("last_change_date"))
            )
    
    # 保留原df1字段和新增的last_change_date列，移除中间别名列
    final_cols = [F.col(f"df1.{field}") for field in df1.columns] + \
                 [F.col(f"last_change_date_{field}") for field in df1.columns if field not in key_fields]
    result_df = result_df.select(final_cols)
    
    return result_df

测试示例 #

from pyspark.sql import SparkSession

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

# 定义列名
df1_cols = ["Name", "Age", "Gender", "Occupation", "Key"]
df2_cols = ["Key", "Name", "Age", "Gender", "Occupation", "last_change_date"]

# 创建DataFrame
df1 = spark.createDataFrame(data_df, schema=df1_cols)
df2 = spark.createDataFrame(data_df2, schema=df2_cols)

# 调用函数，指定Key为关联键
result = compare_dataframes(df1, df2, key_fields=["Key"])
result.show(truncate=False)

说明 #

• 关联时直接使用原关联键字段，避免别名导致的解析错误
• 仅为非关联键字段生成别名，避免列名冲突
• 修正逻辑：当df1与df2对应字段不一致时用当前时间戳，否则沿用df2的last_change_date
• 最终移除中间别名列，只保留原df1字段和新增的last_change_date列

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