Oracle本地服务器到Azure Blob Storage增量/Delta加载管道构建求助

核心思路：增量加载的关键是通过增量标识字段识别Oracle中的新增/更新数据，结合Azure Data Factory（ADF）的变量、Lookup活动维护加载水印，实现每周增量推送CSV到Blob，无需依赖Parquet Delta或Azure SQL。

1. 确认Oracle数据源的增量标识 #

必须有一个能区分新增/更新数据的字段：

• 优先选择时间戳字段：比如create_date（记录数据创建时间）或last_modified_date（记录最后修改时间），字段类型为DATE或TIMESTAMP；如果没有，建议通过Oracle触发器自动新增并维护该字段。
• 若仅需加载新增数据（不处理更新），自增主键ID也可作为增量标识。

本文以last_modified_date时间戳字段为例展开。

2. ADF管道核心配置步骤 #

2.1 初始化增量水印 #

在Blob Storage中创建一个用于存储加载时间戳的文本文件（比如路径watermark/oracle_last_load.txt）：

• 首次运行前，手动写入全量加载的截止时间（例如2024-01-01 00:00:00），后续该文件会自动更新。

2.2 构建增量加载管道 #

步骤1：读取上次加载水印

添加Lookup活动，配置为读取Blob中的watermark/oracle_last_load.txt文件，将读取到的时间戳赋值给管道变量last_load_time。

步骤2：设置当前加载结束时间

添加Set Variable活动，创建字符串类型变量current_load_end_time，赋值为当前UTC时间：

@utcnow()

可根据业务时区调整格式，比如转换为东八区时间：@formatDateTime(convertFromUtc(utcnow(), 'China Standard Time'), 'yyyy-MM-dd HH24:MI:SS')

步骤3：增量复制数据到Blob

添加复制活动，配置如下：

• 源：Oracle数据集，自定义查询语句过滤增量数据：

  SELECT * FROM YOUR_ORACLE_TABLE
  WHERE last_modified_date > TO_TIMESTAMP('@{variables('last_load_time')}', 'YYYY-MM-DD HH24:MI:SS')
    AND last_modified_date <= TO_TIMESTAMP('@{variables('current_load_end_time')}', 'YYYY-MM-DD HH24:MI:SS')
  

  注意：根据Oracle中时间戳字段的实际格式调整TO_TIMESTAMP的格式字符串。
• 目标：Blob Storage数据集，选择CSV格式，建议按加载日期分区存储（比如路径raw_data/your_table/load_date=@{formatDateTime(variables('current_load_end_time'), 'yyyy-MM-dd')}/），便于后续数据管理。
• 复制设置：根据需求选择“追加到文件”或“生成多个文件”模式。

步骤4：更新增量水印

添加Copy活动，将current_load_end_time的值写入Blob的watermark/oracle_last_load.txt文件：

• 源：选择“Inline”数据集，内容设置为@{variables('current_load_end_time')}。
• 目标：指向watermark/oracle_last_load.txt，设置为覆盖写入。

2.3 配置异常分支 #

为避免加载失败时水印被错误更新，需添加Failure分支：

• 在复制活动的“Failure”连接器上，添加失败通知（比如发送邮件），不执行水印更新步骤。

3. 设置每周日触发调度 #

在ADF中创建调度触发器：

• 频率选择“周”，设置每周日的指定时间（比如凌晨2点）触发管道，实现自动增量加载。

关于Parquet Delta的说明 #

Parquet Delta是Delta Lake的格式，主打ACID事务、数据版本控制，适合复杂数据湖场景，但你的需求仅为推送CSV到Blob，完全无需依赖该技术，上述水印+增量查询方案已能满足需求。

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