如何在ADF中实现本地Oracle数据库增量加载及数据变更检测？

一、时间戳字段方案（最易落地） #

• 给Oracle目标表新增更新/创建时间戳字段（比如LAST_UPDATED），用触发器确保数据新增或修改时自动更新该字段：

  CREATE OR REPLACE TRIGGER trg_update_timestamp
  BEFORE INSERT OR UPDATE ON your_table
  FOR EACH ROW
  BEGIN
    :NEW.LAST_UPDATED := SYSTIMESTAMP;
  END;
  /
  

• ADF侧每次运行Copy Activity前，用Lookup Activity读取上一次同步的最大时间戳（可存在ADF变量、Azure SQL小表或Key Vault中）
• 构造Oracle源查询时添加过滤条件：SELECT * FROM your_table WHERE LAST_UPDATED > ?，将上一次的时间戳作为参数传入，拉取增量数据
• 同步完成后，更新存储的时间戳为本次同步的最新时间戳

二、物化视图日志方案 #

• 为需要同步的表创建物化视图日志，专门记录数据变更操作：

  CREATE MATERIALIZED VIEW LOG ON your_table
  WITH PRIMARY KEY, ROWID, SEQUENCE
  INCLUDING NEW VALUES;
  

• ADF直接查询该物化视图日志，通过SNAPTIME$$字段过滤出自上次同步后的变更数据
• 定期清理旧日志避免占用过多存储：PURGE MATERIALIZED VIEW LOG ON your_table;

三、Oracle原生CDC方案（适合大规模多表同步） #

• 启用Oracle原生CDC功能，它会将指定表或全库的变更数据写入专属CDC表
• 先为目标表启用CDC：

  EXEC DBMS_CDC_PUBLISH.CREATE_CHANGE_TABLE(
    owner => 'your_schema',
    change_table_name => 'your_table_ct',
    source_schema => 'your_schema',
    source_table => 'your_table',
    column_type_list => '*',
    capture_values => 'both',
    rs_id => 'y',
    row_id => 'y',
    user_id => 'n',
    timestamp => 'y',
    object_id => 'n',
    source_colmap => 'n',
    target_colmap => 'n',
    options_string => null
  );
  

• ADF通过Oracle连接器读取CDC表，过滤出未同步的变更记录即可

四、自定义变更日志表方案 #

• 新建一张变更日志表（比如your_table_change_log），包含操作类型（INSERT/UPDATE/DELETE）、变更时间、主键、变更内容等字段
• 给目标表添加触发器，每次数据变更时将记录插入日志表：

  CREATE OR REPLACE TRIGGER trg_record_change
  AFTER INSERT OR UPDATE OR DELETE ON your_table
  FOR EACH ROW
  BEGIN
    IF INSERTING THEN
      INSERT INTO your_table_change_log (op_type, change_time, pk_id, new_data)
      VALUES ('INSERT', SYSTIMESTAMP, :NEW.id, :NEW);
    ELSIF UPDATING THEN
      INSERT INTO your_table_change_log (op_type, change_time, pk_id, old_data, new_data)
      VALUES ('UPDATE', SYSTIMESTAMP, :NEW.id, :OLD, :NEW);
    ELSIF DELETING THEN
      INSERT INTO your_table_change_log (op_type, change_time, pk_id, old_data)
      VALUES ('DELETE', SYSTIMESTAMP, :OLD.id, :OLD);
    END IF;
  END;
  /
  

• ADF定期拉取日志表中未同步的记录，同步完成后标记已同步（比如新增IS_SYNCED字段，更新为Y）

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