如何在Synapse数据管道中用ForEach Activity遍历CSV数据集行数据

核心思路 #

Synapse的ForEach Activity确实要求输入是数组格式，但你可以通过简单步骤将CSV数据集转换为可迭代的数组，实现预期的遍历逻辑。

具体操作步骤 #

• 第一步：用Lookup Activity读取CSV数据集
  在ForEach Activity前添加一个Lookup Activity，配置它读取你的两列CSV数据集。Lookup会自动把CSV的每一行转换成JSON对象，最终输出一个包含所有行的数组（格式示例：[{ "property_id": "123", "etag": "abc" }, { "property_id": "456", "etag": "def" }]）。

  注意：必须关闭Lookup Activity的「First row only」选项，才能获取完整的行数据数组。

• 第二步：关联Lookup输出与ForEach输入
  在ForEach Activity的「Items」配置字段中，使用表达式引用Lookup的输出数组：@activity('LookupCSV').output.value（把LookupCSV替换为你实际的Lookup Activity名称）。

• 第三步：在迭代中提取行字段
  在ForEach内部的活动（比如调用API的Web Activity、执行存储过程的Stored Procedure Activity）中，用以下表达式获取当前迭代行的字段：

  • 获取property_id：@item().property_id
  • 获取etag：@item().etag

额外优化提示 #

• 如果CSV数据量较大，Lookup可能出现性能瓶颈，可先用Copy Activity将CSV转换为JSON数组格式的文件，再用Lookup读取该JSON文件；或者结合分页逻辑拆分数据后处理。
• 调用API时建议添加重试机制，避免网络波动导致失败；同时注意API的速率限制，必要时在迭代中加入Wait Activity控制请求频率。

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