Azure Data Factory中For Each活动内数据处理的字段添加与识别问题

一、使用Copy Activity实现（适合快速无复杂转换的场景） #

你可以直接在Copy Activity中通过添加计算列的方式插入处理时间和文件名，步骤如下：

1. 在For Each活动中，确保当前循环的文件信息（文件名）可通过@item().name获取（需保证Get Metadata活动返回的文件列表包含Name属性）。
2. 打开Copy Activity的Mapping标签页：
   • 先完成源文件列与SQL表列的基础映射。
   • 点击添加 → 计算列，新增两个列：
     • 列名：ProcessTime，表达式填写@utcnow()（若需要本地时区，可使用@convertFromUtc(utcnow(), 'Asia/Shanghai')转换）。
     • 列名：FileName，表达式填写@item().name。
   • 将这两个计算列分别映射到Azure SQL表对应的ProcessTime和FileName列。
3. 运行管道后，每个文件的数据会自动带上处理时间和文件名插入SQL表。

二、使用Data Flow Activity实现（适合需要复杂转换的场景） #

先解决文件列无法识别的问题，再添加派生列完成需求：

步骤1：修复文件列识别问题 #

• 确认SFTP数据集的文件格式配置正确：比如CSV文件需指定正确分隔符、是否包含表头；如果是无固定表头的文件，在Data Flow的Source转换中勾选允许架构漂移，并开启推断漂移列。
• 在For Each活动中传递文件名参数给Data Flow：
  • 打开目标Data Flow，添加一个字符串类型的参数（例如fileNameParam）。
  • 在For Each中调用Data Flow时，在参数标签页将fileNameParam的值设为@item().name，同时在Data Flow的Source转换中，将文件路径设置为指向当前处理文件的动态值。

步骤2：添加派生列并写入SQL #

1. 在Data Flow中添加派生列转换：
   • 新增ProcessTime列，使用表达式currentUTC()（若需本地时区，可使用toLocal(currentUTC(), 'Asia/Shanghai')转换）。
   • 新增FileName列，使用表达式$fileNameParam（引用传递进来的文件名参数）。
2. 配置Sink转换：
   • 选择Azure SQL表作为目标，在映射标签页中，将源文件列、派生的ProcessTime和FileName列一一对应到SQL表的列。
3. 运行管道即可完成带额外列的数据写入。

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