创建Data Factory管道实现CSV数据更新插入及缺失补全需求

需求说明 #

现有两个CSV文件：

• 早一周的文件1：含3个月数据，关键列（如报告编号）存在空值，已导入主数据库
• 晚一周的文件2：含3个月数据，与文件1有重叠记录，需基于此完成两个操作：
  1. 识别文件2中的新记录（主库未存在），插入主库
  2. 用文件2的数据补全主库中来自文件1的关键列缺失记录，执行更新

数据示例（中文整理） #

文件1

检查日期	报告编号	患者编号	检查项目
2023/11/23	WD2451	1345	髋关节MRI
2023/11/25		1359	肩关节MRI
2023/11/29		1754	髋关节MRI

文件2

检查日期	报告编号	患者编号	检查项目
2023/11/23	WD2451	1345	髋关节MRI
2023/12/01	WD1983	1359	肩关节MRI
2023/12/04		1754	髋关节MRI

实现步骤 #

1. 前期配置 #

• 定义唯一键：根据业务逻辑确定记录的唯一标识，比如用患者编号+检查项目作为唯一键，避免误判重复记录
• 配置数据集：在ADF中创建文件2的CSV数据集、主数据库（如SQL Server）的数据集

2. 创建映射数据流（核心处理） #

（1）添加源节点

• 源1：连接主数据库表（已导入文件1数据）
• 源2：连接文件2的CSV数据集

（2）关联数据，识别差异

添加查找转换：

• 以源2为左表，关联源1，关联条件设为唯一键匹配，如源2.患者编号 == 源1.患者编号 && 源2.检查项目 == 源1.检查项目
• 输出保留源2的全部字段，以及源1的关键列，如报告编号

（3）拆分处理分支

添加条件拆分，设置两个分支：

• 插入分支：条件为isNull(源1.患者编号)，说明源2的记录在主库中不存在，属于新记录
• 更新分支：条件为!isNull(源1.患者编号) && isNull(源1.报告编号) && !isNull(源2.报告编号)，说明主库已有该记录但关键列缺失，且文件2有补全数据

（4）写入主数据库

• 插入分支：使用SQL接收器，写入模式选择插入，映射所有字段到主库表
• 更新分支：使用SQL接收器，写入模式选择更新，设置更新键为唯一键（患者编号+检查项目），仅映射需要补全的关键列（如报告编号）到主库对应字段

3. 组装管道与触发 #

• 在ADF管道中添加执行数据流活动，调用上述创建的映射数据流
• 设置触发规则：可选择文件到达触发，当文件2上传到存储账户时自动执行，或定时触发，按业务周期执行

注意事项 #

• 唯一键的定义必须贴合实际业务，避免出现重复插入或错误更新
• 若CSV字段名含空格（如原文件的study date），需在数据集配置中调整字段映射，或提前重命名为无空格格式（如study_date）
• 测试时可先使用小批量数据验证分支逻辑，确保插入、更新操作符合预期

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