如何在Databricks中实现Delta表批量加载与增量读取（无CDC属性）

核心方案思路 #

由于无法依赖源表的CDC属性，我们通过时间戳字段过滤或Delta表版本追踪实现首次全量、后续增量的读取逻辑，同时需要维护一个处理状态存储来记录每次处理的截止标记（时间戳/版本号）。

实现步骤与代码示例 #

1. 维护处理状态 #

需要存储每次处理的最后截止标记，推荐用本地JSON文件或Delta状态表，示例用本地文件：

import json

def get_last_processed_mark(mark_type="timestamp"):
    """获取上次处理的截止标记，默认返回初始时间戳"""
    try:
        with open("/data/process_state.json", "r") as f:
            state = json.load(f)
            return state.get(mark_type, "1970-01-01 00:00:00")
    except FileNotFoundError:
        return "1970-01-01 00:00:00" if mark_type == "timestamp" else 0

def update_last_processed_mark(new_mark, mark_type="timestamp"):
    """更新处理状态的截止标记"""
    with open("/data/process_state.json", "w") as f:
        json.dump({mark_type: new_mark}, f)

2. 基于时间戳的增量读取（推荐） #

假设所有Delta表都包含数据写入/更新时自动填充的时间戳字段（如update_time），如果没有需先在ETL写入环节添加该字段：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("DeltaIncrementalLoad").getOrCreate()

# 获取上次处理的时间戳
last_processed_ts = get_last_processed_mark()

# 读取增量数据（首次运行时因初始时间戳极小，自动读取全量）
a_df = spark.table("tablename1").filter(f"update_time > '{last_processed_ts}'")
b_df = spark.table("tablename2").filter(f"update_time > '{last_processed_ts}'")
c_df = spark.table("tablename3").filter(f"update_time > '{last_processed_ts}'")

# 执行表关联（替换为实际业务的关联逻辑）
final_df = spark.sql(f"""
    SELECT 
        a.id, a.col_a, b.col_b, c.col_c
    FROM tablename1 a
    INNER JOIN tablename2 b ON a.id = b.id
    INNER JOIN tablename3 c ON a.id = c.id
    WHERE a.update_time > '{last_processed_ts}'
      AND b.update_time > '{last_processed_ts}'
      AND c.update_time > '{last_processed_ts}'
""")

# 更新处理状态（仅当本次有数据时更新）
if final_df.count() > 0:
    latest_ts = final_df.agg({"update_time": "max"}).collect()[0][0]
    update_last_processed_mark(str(latest_ts))

# 调用内部API传递每行数据
def send_to_api(row):
    """自定义API调用逻辑，需替换为实际接口信息"""
    import requests
    try:
        payload = row.asDict()
        resp = requests.post("http://internal-api:8080/data/ingest", json=payload, timeout=10)
        resp.raise_for_status()
    except Exception as e:
        # 可添加异常日志或重试逻辑
        print(f"Failed to send data {row.id}: {str(e)}")

# 分批处理数据，避免内存溢出
final_df.foreach(send_to_api)

3. 无时间戳时的自定义版本方案 #

如果表没有时间戳字段，可利用Delta表的版本特性实现增量读取：

# 获取上次处理的表版本
last_version_a = get_last_processed_mark(mark_type="version_a")

# 读取Delta表的增量版本数据
a_df = spark.read.format("delta")\
    .option("startingVersion", last_version_a + 1)\
    .table("tablename1")

# 处理完成后更新状态（获取当前表的最新版本）
current_version_a = spark.sql("DESCRIBE HISTORY tablename1")\
    .select("version")\
    .orderBy("version", ascending=False)\
    .first()[0]
update_last_processed_mark(current_version_a, mark_type="version_a")

# 其余表关联、API调用逻辑同时间戳方案

关键注意事项 #

• 时间戳字段必须是数据写入/更新时自动生成，禁止手动修改，否则会导致数据遗漏。
• 使用版本追踪时，需确保Delta表的版本保留策略足够覆盖处理周期，避免版本被清理无法读取增量。
• API调用环节建议添加重试机制和死信队列，处理发送失败的数据，保证数据完整性。

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