如何在Azure Data Factory复制活动中从Oracle动态映射Parquet类型？

核心思路 #

直接利用Oracle系统视图获取表的列元数据（数据类型、精度等），在ADF中通过动态管道将这些元数据映射到Parquet的列类型，完全跳过内容推断，确保类型严格对齐源表定义，同时支持源表架构变更后的自动适配。

具体实现步骤 #

1. 构建表列表配置文件 #

在ADLS或Blob存储中创建JSON格式的配置文件，列出需要同步的Oracle表，示例：

{
  "tables": [
    {"schema": "HR", "name": "EMPLOYEES"},
    {"schema": "HR", "name": "DEPARTMENTS"}
  ]
}

2. 读取配置文件并遍历表列表 #

• 在ADF中创建Lookup活动，读取上述配置文件，获取待同步的表清单。
• 添加For Each活动，遍历Lookup返回的表列表，将每个表的schema和name作为参数传入后续流程。

3. 获取Oracle表的列元数据 #

在For Each内部添加Lookup活动，执行Oracle系统视图查询，获取目标表的列定义：

SELECT 
  COLUMN_NAME, 
  DATA_TYPE, 
  DATA_LENGTH, 
  DATA_PRECISION, 
  DATA_SCALE
FROM ALL_TAB_COLUMNS
WHERE OWNER = '@{item().schema}' 
  AND TABLE_NAME = '@{item().name}'
ORDER BY COLUMN_ID

将查询结果保存为活动输出，后续用于动态映射。

4. 动态生成复制活动的架构映射 #

• 在For Each内部添加复制活动，源为Oracle数据集（需启用动态内容，将表名设置为@{item().schema}.@{item().name}）。
• 目标为ADLS的Parquet数据集，关键配置：
  • 关闭自动映射，利用步骤3获取的元数据，编写动态JSON生成映射规则：

    @json(concat('[', string(join(activity('Get_Oracle_Metadata').output.value, map(item, 
      concat('{"source": {"name": "', item.COLUMN_NAME, '"}, "sink": {"name": "', item.COLUMN_NAME, '", "type": "', 
        if(equals(item.DATA_TYPE, 'VARCHAR2'), 'string', 
        if(equals(item.DATA_TYPE, 'NUMBER'), concat('decimal(', item.DATA_PRECISION, ',', item.DATA_SCALE, ')'),
        if(equals(item.DATA_TYPE, 'DATE'), 'date', 
        if(equals(item.DATA_TYPE, 'TIMESTAMP'), 'timestamp', 
        'string')))), '"}}'
    ))), ']'))
    

    该逻辑会根据Oracle的DATA_TYPE自动映射到Parquet对应类型，可根据实际Oracle类型扩展映射规则。

5. 禁用Parquet的类型推断 #

在Parquet数据集的设置中，将类型推断选项设置为无，确保完全使用动态生成的映射类型，而非从数据内容推断。

6. 适配架构变更 #

当源Oracle表的列类型、新增/删除列时，步骤3的元数据查询会自动获取最新列定义，动态映射同步更新Parquet列类型，无需手动修改ADF管道或数据集。

关键注意事项 #

• 确保ADF的Oracle链接服务有足够权限访问ALL_TAB_COLUMNS系统视图。
• 对于Oracle特殊数据类型（如CLOB、BLOB），需在动态映射中添加对应Parquet类型映射（比如CLOB→string）。
• Parquet数据集文件名可设置为动态值，比如@{item().schema}_@{item().name}.parquet，方便区分不同表的文件。

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