Azure Synapse Pipeline实现XLSX批量转CSV（多工作表拆分）求助

核心流程思路 #

通过嵌套For Each配合Get Metadata活动，实现「遍历所有子文件夹XLSX文件 → 拆分每个文件的工作表 → 按指定命名规则导出CSV」的完整流程。

步骤1：配置存储账户链接服务 #

先创建链接服务（同一存储账户可复用），指向存储XLSX的Azure Blob Storage：

• 类型选择Azure Blob Storage
• 认证优先用托管标识，确保Synapse工作区的托管标识拥有存储账户的Blob数据读取者（源）和Blob数据参与者（目标）权限；也可选择账户密钥认证。

步骤2：构建Pipeline核心组件 #

2.1 定义全局变量（推荐）

在Pipeline中添加以下变量简化表达式：

• 源根路径：字符串类型，填写要遍历的根文件夹路径（如container/source_xlsx/）
• 目标输出路径：字符串类型，填写CSV存储路径（如container/target_csv/）

2.2 获取所有XLSX文件（含子文件夹）

添加Get Metadata活动（命名为Get_All_XLSX）：

• 数据源：选择源存储的Blob数据集，路径设为@variables('源根路径')
• 勾选递归，确保遍历所有子文件夹
• 字段选择Child Items，获取所有文件/文件夹列表

2.3 过滤仅XLSX文件（可选）

添加Filter活动（命名为Filter_XLSX）：

• 输入：@activity('Get_All_XLSX').output.childItems
• 条件：@endsWith(item().name, '.xlsx')，仅保留XLSX格式文件

2.4 遍历每个XLSX文件

添加For Each活动（命名为Loop_XLSX_Files）：

• 迭代项：@activity('Filter_XLSX').output.Value（未加Filter则用@activity('Get_All_XLSX').output.childItems）
• 勾选允许并行执行，可根据文件数量调整并行度（默认20）

在该For Each内部添加Set Variable活动，提取当前文件信息：

• 变量当前文件名：值@split(item().name, '.')[0]（提取不带后缀的文件名）
• 变量当前文件路径：值@item().path（文件完整存储路径）

2.5 获取当前XLSX的工作表列表

添加第二个Get Metadata活动（命名为Get_Sheets）：

• 数据源：选择源存储的Excel数据集，文件路径设为@variables('当前文件路径')
• 字段选择Child Items，获取该文件下所有工作表名称

2.6 遍历工作表并转CSV

添加第二个For Each活动（命名为Loop_Sheets）：

• 迭代项：@activity('Get_Sheets').output.childItems

在该内层For Each中添加Copy Data活动（命名为XLSX_To_CSV）：

源配置：

• 数据集类型：Excel
• 文件路径：@variables('当前文件路径')
• 工作表名称：@item().name（当前遍历到的工作表）
• 按需勾选第一行作为标题

目标配置：

• 数据集类型：CSV
• 文件路径：@concat(variables('目标输出路径'), variables('当前文件名'), '_', item().name, '.csv')
• CSV属性：分隔符选逗号，编码设为UTF-8，按需调整其他格式选项

关键注意事项 #

• 权限校验：确保Synapse对源存储有读取权限、对目标存储有写入权限，避免访问被拒
• 文件名兼容：若源XLSX文件名含特殊字符（如空格、斜杠），需提前处理，避免CSV文件名不符合存储规范
• 性能优化：文件数量极大时，可降低For Each并行度，避免触发存储账户限流

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