如何从SCD1 Delta Live Tables获取更新/插入记录及增量数据集？

一、操作日期与类型的内部列说明 #

Delta Live Tables（DLT）基于CDC实现SCD1时，默认不会自动添加单条记录级的操作类型（INSERT/UPDATE）和操作日期内部列。你可以通过两种方式获取这类元数据：

• 保留CDC源自带元数据：如果你的CDC源（如Debezium、各类消息队列）本身包含操作类型（比如op字段）、操作时间戳（比如ts_ms），在DLT的CDC处理逻辑中直接将这些字段保留到目标SCD1表即可。
• 利用Delta事务日志/历史：Delta表的事务日志（_delta_log目录）会记录批次级的操作时间和类型，但直接解析日志成本较高。可以用DESCRIBE HISTORY <table-name>命令查看表的变更批次历史，或通过deltaTable.history() API获取批次元数据，但这种方式只能拿到批次维度的信息，无法关联到单条记录的操作类型。

二、获取增量数据集并发送至AWS SQS #

要捕获SCD1表的新增/更新增量数据并发送到SQS，可按以下步骤操作：

1. 提取增量数据
   基于Delta表的版本或时间戳对比，获取两次处理间隔内的增量记录：

   from delta.tables import DeltaTable
   
   # 加载目标SCD1表
   scd1_table = DeltaTable.forPath(spark, "/path/to/your-scd1-table")
   # 从外部存储（如Redis、数据库）获取上一次处理的版本号
   last_processed_version = 3
   # 获取当前表的最新版本
   current_version = scd1_table.history(1).select("version").collect()[0][0]
   
   # 提取增量数据（新增+更新的记录）
   incremental_df = spark.sql(f"""
       SELECT * FROM delta.`/path/to/your-scd1-table` VERSION AS OF {current_version}
       EXCEPT
       SELECT * FROM delta.`/path/to/your-scd1-table` VERSION AS OF {last_processed_version}
   """)
   

2. 发送至AWS SQS
   使用boto3 SDK批量发送增量数据到SQS，避免单条发送的性能损耗：

   import boto3
   import json
   
   sqs_client = boto3.client('sqs', region_name='your-region-id')
   queue_url = 'your-sqs-queue-url'
   
   # 将DataFrame转换为JSON列表，按SQS批量限制拆分
   records = [json.dumps(row.asDict()) for row in incremental_df.collect()]
   batch_size = 10  # SQS单批次最多支持10条消息
   for i in range(0, len(records), batch_size):
       batch = records[i:i+batch_size]
       entries = [{'Id': str(idx), 'MessageBody': msg} for idx, msg in enumerate(batch)]
       sqs_client.send_message_batch(QueueUrl=queue_url, Entries=entries)
   

三、表血缘关系排查 #

DLT表的血缘关系需要在Databricks工作区的对应模块查看，若无法找到可能是以下原因：

• 目标表是临时视图而非物化Delta表：临时对象不会被记录血缘，需确保DLT流水线生成的是持久化的Delta表。
• 流水线未启用血缘采集：检查DLT流水线设置中的“数据治理”选项，确认已开启血缘采集开关。
• 流水线未成功运行：只有完成至少一次成功运行的流水线，其生成的表才会在数据洞察/数据治理模块中展示血缘关系。

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