如何从Spark SQL DataFrame增量写入数据到ADLS Gen2（Delta格式）

要实现首次全量、后续增量写入ADLS Gen2的Delta表，核心思路是先检测目标表是否存在，不存在则全量初始化，存在则通过Delta的MERGE操作实现增量合并（更新已有数据、插入新数据）。以下是具体实现：

关键前提 #

你的源数据表（datavault.tablename）需要有唯一标识字段（比如主键ID）或增量标识字段（比如最后更新时间戳update_ts），用于匹配目标表中的数据，判断是插入新数据还是更新已有数据。

代码实现 #

from delta.tables import DeltaTable

# 定义目标表名和ADLS存储路径
target_table_name = "tablename"
target_path = "/mnt/storagelocation/tablename"

# 读取源数据（建议根据增量逻辑优化查询，比如只取上次同步后的新数据）
# 示例增量查询：WHERE update_ts > (SELECT COALESCE(MAX(update_ts), '1970-01-01') FROM tablename)
source_df = spark.sql("SELECT * FROM datavault.tablename")

# 检查目标Delta表是否存在
if spark.catalog.tableExists(target_table_name):
    # 加载目标Delta表
    delta_table = DeltaTable.forName(spark, target_table_name)
    
    # 执行MERGE操作：根据唯一键匹配，匹配到则更新，没匹配到则插入
    # 替换`id`为你的实际唯一标识字段，可自定义更新/插入的字段列表
    delta_table.alias("target") \
        .merge(
            source_df.alias("source"),
            "target.id = source.id"  # 匹配条件，根据业务调整
        ) \
        .whenMatchedUpdateAll()  # 匹配到则更新所有字段，也可指定具体字段
        .whenNotMatchedInsertAll()  # 未匹配到则插入所有字段
        .execute()
else:
    # 首次执行：全量写入并创建Delta表
    source_df.write \
        .format("delta") \
        .mode("overwrite") \
        .option("mergeSchema", "true")  # 自动兼容Schema变化
        .option("path", target_path) \
        .saveAsTable(target_table_name)

重要细节说明 #

• 增量数据过滤：如果源表数据量较大，不要每次全量读取，建议在spark.sql中加入过滤条件（比如基于上次同步的时间戳/最大主键），只获取新增或变更的数据，大幅提升性能。
• Schema处理：mergeSchema选项支持自动合并源表与目标表的Schema差异（比如新增字段），如果涉及字段类型变更，需确保类型兼容或提前处理Schema。
• 原代码修正：你提供的代码存在笔误test..write，需修正为test.write。
• ADLS权限：确保Spark集群已正确挂载ADLS Gen2路径，且拥有对应读写权限。

可选优化 #

• 维护同步日志表，记录每次同步的时间戳或最大主键值，下次同步时基于日志精准过滤源数据。
• 对大规模Delta表执行OPTIMIZE和ZORDER BY操作，优化后续查询性能。

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