ADF映射数据流处理可变未知Schema的CSV文件映射问题求助

核心思路 #

先提取当前CSV的所有列名，关联映射表筛选出有效映射关系，再动态重命名列后下沉到目标SQL表。

具体实现步骤 #

1. 启用Schema漂移并提取CSV列元数据 #

• 在CSV源数据集开启允许Schema漂移，勾选导入未知列。
• 在映射数据流的源转换后添加派生列，用表达式 array(keys()) 生成当前CSV的全列名数组，命名为csv_columns。
• 添加聚合转换，将csv_columns聚合为单一行的数组（因每行列名一致，取聚合后第一个值即可），输出仅含完整列名数组的单行数据。

2. 关联SQL映射表筛选有效映射 #

• 添加新源连接到SQL Server的列映射表（建议结构：source_column VARCHAR, target_column VARCHAR）。
• 用交叉连接将CSV列元数据与映射表关联，再添加筛选转换，用contains(csv_columns, source_column)过滤出当前CSV实际存在的源列映射关系。

3. 生成动态列重命名规则 #

• 添加脚本转换，用以下表达式生成键值对形式的重命名映射数组，命名为rename_mappings：

  map(collect(
      struct(
          source_column as fromName,
          target_column as toName
      )
  ))
  

4. 动态重命名列并下沉 #

• 返回CSV源的主数据流分支，添加派生列转换，用表达式 fold(rename_mappings, {}, (acc, map) => merge(acc, byName(map.fromName) as map.toName)) 完成列的动态重命名。
• 添加下沉转换连接到目标SQL表，开启允许Schema漂移，设置为仅写入匹配的列，仅同步映射表中存在的字段。

关键注意事项 #

• 映射表需覆盖所有可能的CSV源列与SQL目标列的对应关系，未匹配列会被自动忽略。
• 确保CSV源列与SQL目标列的数据类型兼容，避免同步时出现类型错误。
• 循环处理单文件时，上述流程会自动适配每个CSV的Schema，无需提前知晓具体列名。

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