Databricks Autoloader导入S3 CSV到Delta表的重复数据处理问题

问题根因 #

Autoloader默认仅基于文件路径、修改时间等元数据标记已处理文件，不会校验文件内容或业务数据的重复性。你上传的两个文件名不同、内容相同的文件会被判定为全新待处理文件，全量写入Delta表，最终出现数据翻倍。

推荐解决方案 #

方案1：基于业务主键幂等写入（通用，适用绝大多数场景） #

通过Delta Lake的MERGE操作实现幂等写入，业务主键匹配时更新旧数据、不匹配时插入新数据，从根本上避免业务维度的数据重复。你可以结合Structured Streaming的foreachBatch算子实现流式merge：

1. 首先定义Upsert逻辑函数

def upsert_to_delta(batch_df, batch_id):
    target_table_path = "s3://some-s3-path/spark_stream_processing/target/"
    # 注册当前批次数据为临时视图
    batch_df.createOrReplaceTempView("current_batch")
    
    # 首次运行表不存在时先建表，后续运行走merge逻辑
    if not DeltaTable.isDeltaTable(spark, target_table_path):
        batch_df.write.format("delta").save(target_table_path)
        return
    
    # 执行merge操作，此处假设id为唯一业务主键，联合主键可修改ON后的匹配条件
    spark.sql(f"""
        MERGE INTO delta.`{target_table_path}` t
        USING current_batch s
        ON t.id = s.id
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)

2. 修改你的流处理代码，调用上述函数

from delta.tables import DeltaTable

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .option("header", True) \
  .schema("id string,name string, age string,city string") \
  .load("s3://some-s3-path/source/") \
  .writeStream \
  .option("checkpointLocation", "s3://some-s3-path/tgt_checkpoint_0928/") \
  .option("mergeSchema", "true") \
  .foreachBatch(upsert_to_delta) \
  .start()

方案2：基于文件内容去重（仅适用相同内容文件重复上传场景） #

如果你只需要过滤掉内容完全一致的重复文件，不需要处理单条业务数据重复的场景，可以开启Autoloader的文件哈希校验逻辑，内容哈希相同的文件哪怕文件名不同也会被跳过：

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .option("header", True) \
  .option("cloudFiles.fileHashMode", "md5") \
  .schema("id string,name string, age string,city string") \
  .load("s3://some-s3-path/source/") \
  .writeStream.format("delta") \
  .option("mergeSchema", "true") \
  .option("checkpointLocation", "s3://some-s3-path/tgt_checkpoint_0928/") \
  .start("s3://some-s3-path/spark_stream_processing/target/")

方案3：简单去重写入（适用小数据量、无数据更新需求场景） #

如果不需要更新旧数据，仅需要保证主键唯一，也可以在写入前对批次数据和已有表数据做去重后再追加：

def deduplicate_write(batch_df, batch_id):
    target_path = "s3://some-s3-path/spark_stream_processing/target/"
    # 对当前批次按主键去重
    dedup_batch = batch_df.dropDuplicates(["id"])
    # 过滤掉已经存在于目标表的主键数据
    if DeltaTable.isDeltaTable(spark, target_path):
        existing_ids = spark.read.format("delta").load(target_path).select("id")
        dedup_batch = dedup_batch.join(existing_ids, on="id", how="left_anti")
    # 追加写入去重后的数据
    dedup_batch.write.format("delta").mode("append").save(target_path)

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