咨询：在Databricks中用PySpark DataFrame优化CSV文件逐行对比方案

核心思路 #

通过姓名、姓氏、出生日期这类标识字段将两个DataFrame做关联，逐字段对比同一标识下的行值，提取并结构化展示行内差异，同时可单独处理两边独有的行。

具体实现步骤（PySpark） #

1. 加载CSV文件到DataFrame
   假设两个数据库导出的对应CSV为source.csv和target.csv，先加载并统一字段名（确保字段完全对齐）：

   df_source = spark.read.csv("/path/to/source.csv", header=True, inferSchema=True)
   df_target = spark.read.csv("/path/to/target.csv", header=True, inferSchema=True)
   
   # 若导出字段名不一致，手动对齐
   df_target = df_target.toDF(*df_source.columns)
   

2. 定义标识字段与对比字段
   指定用于匹配行的唯一标识，以及需要对比的其他字段：

   identifier_cols = ["first_name", "last_name", "date_of_birth"]
   compare_cols = [col for col in df_source.columns if col not in identifier_cols]
   

3. 关联两个DataFrame并区分来源
   用inner join关联同标识的行，给字段加前缀区分来源：

   df_joined = df_source.alias("src")\
       .join(df_target.alias("tgt"), on=identifier_cols, how="inner")
   

4. 生成差异检测逻辑
   遍历对比字段，生成差异标记列，同时记录差异的具体值：

   from pyspark.sql.functions import col, when, struct, lit, concat_ws
   
   diff_cols = []
   for c in compare_cols:
       # 处理空值：null与null视为相等，否则对比值
       is_diff = when((col(f"src.{c}").isNull() & col(f"tgt.{c}").isNull()), False)\
                   .otherwise(col(f"src.{c}") != col(f"tgt.{c}"))
       # 结构化记录差异详情
       diff_detail = struct(
           lit(c).alias("field"),
           col(f"src.{c}").alias("source_value"),
           col(f"tgt.{c}").alias("target_value")
       )
       diff_cols.append(when(is_diff, diff_detail).alias(f"diff_{c}"))
   
   # 合并所有差异为可读格式，过滤无差异行
   df_diff = df_joined.select(
       *identifier_cols,
       concat_ws(" | ", [col(f"diff_{c}").cast("string") for c in compare_cols]).alias("all_diffs")
   ).filter(col("all_diffs") != "")
   

5. 处理两边独有的行
   若需找出仅在某一个DataFrame存在的行，单独用left anti join：

   # 仅source中存在的行
   source_only = df_source.join(df_target, on=identifier_cols, how="left_anti")
   # 仅target中存在的行
   target_only = df_target.join(df_source, on=identifier_cols, how="left_anti")
   

注意事项 #

• 字段类型对齐：确保同名字段类型一致（比如日期字段统一为DateType，避免字符串格式差异导致误判）
• 空值处理：Spark中null == null返回null，必须单独处理空值相等的场景
• 性能优化：数据量较大时，先对标识字段分区或缓存，提升join效率

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