Azure Synapse Pipeline中动态设置Parquet列名并合并CSV的方案咨询

你的思路是可行的，且可以通过列名列表完成动态映射，以下是具体实现步骤和注意事项：

一、获取带表头文件的列名数组 #

1. 创建数组变量：在Pipeline中新建一个数组类型变量（如columnNames），用于存储提取的列名。
2. Lookup Activity配置：
   • 指向目标带表头CSV文件，对应数据集需开启First row as header = True。
   • 在Lookup Activity设置中选择First row only，仅读取文件的表头行。
3. Set Variable Activity配置：
   • 将Lookup输出转换为列名数组，赋值给columnNames变量，使用表达式：

     @keys(activity('Lookup_Header_File').output.firstRow)
     

   该表达式会提取Lookup结果中第一行的所有键（即CSV列名），生成列名数组。

二、用Copy Activity合并并映射文件 #

方案1：分两个Copy Activity处理不同类型文件（更可靠） #

由于带表头文件需要跳过第一行（表头行），不带表头文件直接读取所有行，分开处理更清晰：

处理带表头的文件

• 源数据集：关闭First row as header，设置Skip line count = 1（跳过表头行）。
• 映射配置：切换到Copy Activity映射视图的JSON编辑器，替换为以下表达式生成映射规则：

  {
      "type": "TabularTranslator",
      "columnMappings": "@reduce(variables('columnNames'), '', (acc, item) => concat(acc, if(empty(acc), '', ','), string(indexOf(variables('columnNames'), item)+1), ':', item))"
  }
  

  该表达式会生成"1:列名1,2:列名2,..."的映射字符串，将文件第1、2...列对应到提取的列名上。
• 目标数据集：设置为合并目标（如Blob存储、Synapse表），确保Schema与列名数组匹配。

处理不带表头的文件

• 源数据集：关闭First row as header，设置Skip line count = 0。
• 映射配置：使用与上述完全相同的动态映射表达式，直接将文件列对应到提取的列名上。
• 目标数据集：选择与上述相同的目标，设置为追加写入模式。

方案2：单个Copy Activity处理所有文件（需文件名规律支持） #

如果想用一个Copy Activity处理所有文件，需依赖文件名规律区分类型（如带表头文件名包含_with_header）：

• 源使用Wildcard路径，结合Filter筛选文件。
• 在Copy Activity源设置中，通过动态内容设置skipLineCount：

  @if(contains(item().name, '_with_header'), 1, 0)
  

  注意：需将源切换为File list模式，并在Additional properties中启用skipLineCount的动态赋值。
• 映射部分同样使用方案1中的动态映射表达式。

三、关键注意事项 #

• 确保所有文件的列数与带表头文件的列数一致，否则会出现映射错误。
• 目标数据集建议开启Schema drift，避免因列名或顺序变化导致失败。
• 若合并到同一个文件，目标数据集需设置固定文件名，并选择Append写入模式。

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