如何避免Delta Lake重复Parquet文件引发Azure Data Explorer表数据重复

一、在Synapse端优化：避免无意义的文件生成 #

1. 先校验数据变更再执行Merge #

当前代码无论是否有实际数据变更都会执行Merge操作，进而生成冗余Parquet文件。可以先筛选出真正需要更新/插入的记录，仅当存在有效变更时才执行Merge：

from pyspark.sql import functions as F
from delta.tables import DeltaTable

delta_table_path = 'abfss://test@adlstest.dfs.core.windows.net/db/tests'
gold_database_name = "your_gold_db" # 替换为实际库名

table_exists = DeltaTable.isDeltaTable(spark, delta_table_path)

if table_exists:
    deltaTable = DeltaTable.forPath(spark, delta_table_path)
    delta_df = deltaTable.toDF()
    
    # 筛选出需要更新的记录：Key匹配且旧值小于新值
    update_candidates = df_final.join(delta_df, on="Key", how="inner")\
                               .filter(delta_df.Value < df_final.Value)
    # 筛选出需要插入的记录：新Key不存在于现有表中
    insert_candidates = df_final.join(delta_df, on="Key", how="left_anti")
    
    # 合并所有需要变更的数据
    change_df = update_candidates.select(df_final["*"]).union(insert_candidates)
    
    if change_df.count() > 0:
        deltaTable.alias("delta").merge(
            change_df.alias("updates"),
            "delta.Key= updates.Key") \
        .whenMatchedUpdate(
            condition = "delta.Value < updates.Value",
            set = {"Value": "updates.Value"}) \
        .whenNotMatchedInsertAll() \
        .execute()
    else:
        print("无有效数据变更，跳过Merge操作")
else:
    (df_final.write.option("header", True)
                    .partitionBy("CITYID", "month")
                    .mode("append")
                    .format("delta")
                    .option("mergeSchema", "true")
                    .saveAsTable(f"{gold_database_name}.city")
                    )

2. 定期清理Delta旧版本文件 #

Delta Lake会保留历史版本的Parquet文件，可通过optimize合并小文件、vacuum清理过期历史文件，减少ADX读取到旧数据的概率：

# 在Merge执行完成后添加以下代码
deltaTable.optimize().execute()
# 保留24小时的旧版本（可根据业务调整时长，注意避免清理仍在被读取的文件）
deltaTable.vacuum(24)

二、在ADX端适配Delta事务机制（推荐） #

ADX直接读取底层Parquet文件时无法识别Delta的事务日志，会读取所有历史版本文件导致重复。优先采用以下原生支持方案：

1. 创建ADX Delta Lake外部表 #

ADX已支持直接解析Delta Lake的事务日志，仅加载最新版本的数据。修改你的Kusto命令：

.create external table TEST (CITYID: string, Key: string, Value: string)
kind=delta
dataformat=delta
(
    h@'abfss://test@adlstest.dfs.core.windows.net/db/tests;impersonate'
)

此方式无需修改Synapse代码，ADX会自动处理Delta的版本管理，从根源避免重复数据。

2. 同步Delta数据到ADX内部表 #

如果无法使用Delta外部表，可通过Synapse Pipeline的Copy Activity，或ADX的ingest from delta命令，将Delta表的最新数据同步到ADX内部表，替代直接读取底层存储的方式。

方案总结 #

优先选择ADX Delta外部表方案，无需改动Synapse业务逻辑，直接利用ADX对Delta的原生支持；如果受限于ADX版本，再结合Synapse端的变更校验+旧文件清理组合方案解决问题。

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