Azure SQL 200张表CDC数据导出至Blob Storage单表单CSV的ADF管道求助

1. 获取CDC表清单 #

• 在Azure SQL中执行查询，提取所有启用CDC的表信息：

  SELECT s.name AS schema_name, t.name AS table_name
  FROM sys.tables t
  JOIN sys.schemas s ON t.schema_id = s.schema_id
  WHERE is_tracked_by_cdc = 1
  

• 在ADF中添加Lookup活动，执行上述SQL，取消勾选「First row only」，确保返回全部目标表的架构和名称。

2. 配置表遍历逻辑 #

• 添加ForEach活动，将「Items」设置为Lookup活动的输出结果：

  @activity('Lookup CDC Tables').output.value
  

• 根据SQL和Blob的性能承载能力，选择是否开启「Sequential」（串行执行），并发场景可关闭该选项并调整并发数。

3. 参数化源数据集（Azure SQL） #

• 创建Azure SQL数据集，新增两个参数：schemaName、tableName
• 将数据集的「Table」属性绑定为：

  @{dataset().schemaName}.@{dataset().tableName}
  

• 在复制活动的源配置中，开启「增量复制」，选择对应CDC模式（基于时间或CDC函数），按需设置起始/结束时间范围，或直接调用CDC内置函数（如cdc.fn_cdc_get_all_changes_<capture_instance>）。

4. 参数化Sink数据集（Blob Storage） #

• 创建Blob Storage数据集，新增参数：tableName
• 将数据集的「文件路径」设置为：

  @{concat('cdc-exports/', dataset().tableName, '.csv')}
  

• 配置CSV格式属性，包括分隔符、引号规则、编码等。

5. 循环内配置复制活动 #

• 在ForEach活动内部添加复制活动：
  • 源端选择参数化后的Azure SQL数据集，传入当前循环项的参数：
    • schemaName: @item().schema_name
    • tableName: @item().table_name
  • Sink端选择参数化后的Blob数据集，传入参数：
    • tableName: @item().table_name
  • 配置字段映射：若表结构统一可使用自动映射，结构差异场景需配置动态映射或提前规范CDC字段格式。

6. 增量同步优化（可选） #

• 在Azure SQL中创建同步状态表，存储schema_name、table_name、last_sync_time
• 新增Lookup活动读取各表上次同步时间，作为复制活动CDC范围的起始时间
• 同步完成后，添加存储过程活动，更新状态表的last_sync_time为当前时间

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