Databricks基于最新CDF向含AutoIncrement Identity的表插入增量变更

Databricks没有内置托管CDF偏移量的开箱即用特性，手动维护偏移量+原生CDF读取能力就能实现 exactly-once 增量写入，完全不会和目标表的AutoIncrement自增列冲突，也不需要每次全量重跑。

1. 前置准备：建轻量偏移量维护表 #

只需要一张数据量极小的Delta表记录每个管道上次处理成功的源表版本即可，建表语句：

CREATE TABLE IF NOT EXISTS main.pipeline.cdf_offset (
  pipeline_name STRING COMMENT '管道唯一标识，用于区分不同任务',
  last_processed_version LONG COMMENT '上次成功处理完成的源Delta表版本号',
  last_processed_ts TIMESTAMP COMMENT '上次处理完成时间'
) USING DELTA
TBLPROPERTIES ('delta.autoOptimize.optimizeWrite' = 'true');

-- 首次运行初始化偏移量，版本设为源表开启CDF前的最后一个版本即可，全新空表直接填-1
INSERT INTO main.pipeline.cdf_offset 
VALUES ('csv_autoloader_to_target', -1, current_timestamp());

2. 增量任务核心逻辑 #

每次任务启动先读偏移量，拉取对应版本区间的新增CDF数据，处理后写入目标表，成功后更新偏移量即可，PySpark示例代码：

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

# 读取上次处理的偏移量
offset_record = spark.sql("""
    SELECT last_processed_version 
    FROM main.pipeline.cdf_offset 
    WHERE pipeline_name = 'csv_autoloader_to_target'
""").collect()[0]
last_processed_version = offset_record["last_processed_version"]

# 获取源表现状：最新版本、CDF最早可查询版本
source_tbl = DeltaTable.forName(spark, "main.default.autoloader_csv_source")
current_version = source_tbl.history().select(F.max("version")).first()[0]
earliest_available_cdf_version = source_tbl.history().select(F.min("version")).first()[0]

# 无新数据直接退出
if current_version <= last_processed_version:
    print(f"无新增增量，当前已处理版本：{last_processed_version}，源表最新版本：{current_version}")
    exit(0)

# CDF日志过期时触发全量兜底（CDF默认留存30天，可通过表属性修改留存周期）
if last_processed_version < earliest_available_cdf_version:
    print(f"版本{last_processed_version}对应的CDF日志已清理，触发全量重跑兜底")
    # 此处直接复用你已经验证可用的全量写入逻辑
    full_df = spark.table("main.default.autoloader_csv_source")
    processed_full_df = full_df # 替换为你的字段处理逻辑
    processed_full_df.write.format("delta").mode("append").saveAsTable("main.default.target_table_with_identity")
    # 更新偏移量到当前版本
    spark.sql(f"""
        UPDATE main.pipeline.cdf_offset
        SET last_processed_version = {current_version}, last_processed_ts = current_timestamp()
        WHERE pipeline_name = 'csv_autoloader_to_target'
    """)
    exit(0)

# 正常读取增量数据：只取两个版本区间内的insert类型记录，就是Auto Loader新导入的CSV数据
incremental_df = spark.read.format("delta") \
    .option("readChangeFeed", "true") \
    .option("startingVersion", last_processed_version + 1) \
    .option("endingVersion", current_version) \
    .table("main.default.autoloader_csv_source") \
    .filter("_change_type = 'insert'") \
    .drop("_change_type", "_commit_version", "_commit_timestamp") # 删掉CDF自带的元数据列

# 此处添加你自定义的字段清洗、转换逻辑
processed_inc_df = incremental_df

# 写入带自增列的目标表：用append模式即可，自增ID会自动生成，不要用overwrite避免自增序列重置
processed_inc_df.write.format("delta") \
    .mode("append") \
    .saveAsTable("main.default.target_table_with_identity")

# 【关键】写入成功后再更新偏移量，任务失败时偏移量不变，下次重跑不会丢数据
spark.sql(f"""
    UPDATE main.pipeline.cdf_offset
    SET last_processed_version = {current_version}, last_processed_ts = current_timestamp()
    WHERE pipeline_name = 'csv_autoloader_to_target'
""")

print(f"增量处理完成，覆盖版本区间：{last_processed_version + 1} ~ {current_version}")

常见避坑点 #

• 偏移量更新一定要放在目标表写入成功之后，不要提前更新，否则任务失败会导致数据漏写
• 带AutoIncrement自增列的目标表只能用append模式写入，overwrite会重置自增序列，导致ID重复
• 如果需要修改CDF日志留存周期，可以给源表设置表属性delta.logRetentionDuration，比如设为90天就执行ALTER TABLE main.default.autoloader_csv_source SET TBLPROPERTIES ('delta.logRetentionDuration' = 'interval 90 days')
• 多任务并发跑同一管道时，给偏移量表的更新加乐观锁校验，避免并发更新导致偏移量错乱

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