Azure Data Factory新手实操：如何创建管道将CSV数据导入含外键关联的CustomerProduct表

作为ADF新手，你遇到的核心问题是需要将CSV中的业务字段（CustomerNumber、ErpNumber）关联到对应主表的主键（CustomerId、ProductId），再插入到CustomerProduct关联表中。下面给你两种最实用的实现方案，适合不同场景：

方案一：使用Data Flow（推荐，可视化操作，适合新手） #

Data Flow支持可视化的关联/转换操作，能直观完成字段映射和数据转换，步骤如下：

1. 创建Data Flow活动
   在你的管道中添加一个Data Flow活动，点击创建新的数据流。

2. 添加CSV源
   在数据流画布中，添加源转换，选择你之前创建的Blob CSV数据集，确保CSV的列（CustomerNumber、ErpNumber、Name）被正确识别。

3. 关联Customer表获取CustomerId
   添加Lookup转换，连接到你的SQL链接服务，选择Customer表作为Lookup源，设置关联条件：

   source.CustomerNumber == customer.CustomerNumber
   

   勾选需要返回的字段：Id（重命名为CustomerId）。

4. 关联Product表获取ProductId
   再添加一个Lookup转换，连接到SQL链接服务，选择Product表，设置关联条件：

   source.ErpNumber == product.ErpNumber
   

   勾选返回字段：Id（重命名为ProductId）。

5. 派生必要字段
   添加派生列转换，生成CustomerProduct表需要的其他字段：

   • Id: 使用uuid()函数生成唯一标识符
   • CreatedOn: 使用currentTimestamp()生成当前时间
     （如果有其他默认字段，也可以在这里设置固定值或表达式）

6. 设置接收器（目标表）
   添加接收器转换，选择你之前创建的CustomerProduct SQL数据集，设置写入行为为插入，然后完成字段映射（将派生的Id、CustomerId、ProductId，以及源的Name、CreatedOn对应到目标表的列）。

7. 运行管道
   保存数据流和管道，触发运行即可完成数据导入。

方案二：Lookup活动 + 存储过程（适合复杂业务逻辑场景） #

如果需要更灵活的业务逻辑处理（比如异常判断、自定义日志），可以用Lookup获取主表映射，再通过存储过程完成插入：

1. 创建两个Lookup活动获取映射

   • 在管道中添加第一个Lookup活动，连接到SQL链接服务，执行查询：

     SELECT CustomerNumber, Id AS CustomerId FROM dbo.Customer
     

     这个活动会返回所有客户的CustomerNumber和CustomerId映射。
   • 添加第二个Lookup活动，执行查询：

     SELECT ErpNumber, Id AS ProductId FROM dbo.Product
     

     返回产品的ErpNumber和ProductId映射。

2. 创建插入数据的存储过程
   在你的SQL Server中创建存储过程，用于接收CSV行数据并完成关联插入：

   CREATE PROCEDURE dbo.InsertCustomerProduct
       @CustomerNumber NVARCHAR(50),
       @ErpNumber NVARCHAR(50),
       @Name NVARCHAR(255)
   AS
   BEGIN
       SET NOCOUNT ON;
       
       -- 处理不存在的CustomerNumber或ErpNumber的情况（可选，根据需求调整）
       DECLARE @CustomerId UNIQUEIDENTIFIER = (SELECT Id FROM dbo.Customer WHERE CustomerNumber = @CustomerNumber);
       DECLARE @ProductId UNIQUEIDENTIFIER = (SELECT Id FROM dbo.Product WHERE ErpNumber = @ErpNumber);
       
       IF @CustomerId IS NOT NULL AND @ProductId IS NOT NULL
       BEGIN
           INSERT INTO dbo.CustomerProduct (Id, CustomerId, ProductId, Name, CreatedOn)
           VALUES (NEWID(), @CustomerId, @ProductId, @Name, GETDATE());
       END
       ELSE
       BEGIN
           -- 可选：记录错误日志，比如插入到错误表
           INSERT INTO dbo.CustomerProduct_ErrorLog (CustomerNumber, ErpNumber, ErrorMessage)
           VALUES (@CustomerNumber, @ErpNumber, 'Invalid CustomerNumber or ErpNumber');
       END
   END
   

3. 配置复制活动调用存储过程

   • 添加复制活动，源选择你的CSV数据集，目标选择SQL数据集。
   • 在目标设置中，选择写入行为为“存储过程”，指定刚才创建的InsertCustomerProduct存储过程。
   • 进入“映射”标签，将CSV的CustomerNumber、ErpNumber、Name列分别映射到存储过程的对应参数。

注意事项 #

• 异常处理：如果CSV中存在不存在的CustomerNumber或ErpNumber，Data Flow中可以通过“允许拒绝行”功能将错误行输出到Blob存储；存储过程方案中可以添加错误日志逻辑。
• 性能优化：如果Customer或Product表数据量很大，Lookup活动可能会有性能问题，此时优先选择Data Flow方案，它支持分布式处理。
• 字段映射验证：确保所有目标表的必填字段都被正确赋值，比如Id和CreatedOn。

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