Azure Data Factory中根据Blob文件名匹配插入SQL表并实现upsert的方法

前置准备 #

• 提前创建好Azure存储账户的链接服务、目标SQL数据库的链接服务
• 确认所有目标SQL表已经提前创建完成，表结构与对应CSV的字段对齐，且每张表都已设置好唯一标识字段（用于Upsert匹配）

步骤1：配置Get Metadata活动获取所有源文件 #

• 新建管道，添加Get Metadata活动，关联指向源存储文件夹的CSV数据集
• 在Get Metadata的「字段列表」中添加Child Items选项，执行后会返回文件夹下所有文件名的数组

步骤2：添加ForEach活动遍历所有文件 #

• 把Get Metadata的输出作为ForEach的输入，ForEach的「项」配置表达式：@activity('Get Metadata').output.childItems
• 进入ForEach活动内部，后续所有操作都在ForEach循环内执行

步骤3：解析文件名生成目标表名 #

在ForEach内部先添加一个设置变量活动，提前创建字符串类型变量targetTableName，变量值填如下动态表达式：

@join(take(split(replace(item().name, '.csv', ''), '_'), add(length(split(replace(item().name, '.csv', ''), '_')), -1)), '_')

该表达式会自动去掉CSV后缀，同时删除文件名末尾的日期段，拼接生成符合要求的目标表名，比如156468a5s5s54_Inventory_08022021.csv处理后会得到156468a5s5s54_Inventory。

步骤4：配置复制活动实现Upsert #

• 在ForEach内部添加复制活动，源数据集选择之前的CSV数据集，配置动态文件名参数为@item().name，每次循环读取当前遍历到的文件
• 接收器数据集选择SQL数据集，配置动态表名参数为@variables('targetTableName')，自动匹配到对应的目标SQL表
• 进入复制活动的「接收器」设置，写入行为选择Upsert
• 配置Upsert的键列：选择表中提前预设的唯一标识字段，ADF会自动根据该字段判断匹配行，存在则更新、不存在则插入

注意事项 #

• 如果CSV的列名和SQL表的列名不完全一致，需要在复制活动的「映射」标签页手动配置字段映射，字段名完全一致时可直接开启自动映射
• 如果待处理文件数量很大，可以开启ForEach的「批量并行」选项提高处理效率，并行数不要超过ADF配额限制，同时不要超过目标SQL库的并发连接上限
• 可以在管道开头添加过滤活动，筛掉非CSV格式的文件，避免执行报错

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