如何在 Azure 数据工厂中为与外键相关的表创建管道并执行复制活动？

Posted 2023-03-28

【中文标题】如何在 Azure 数据工厂中为与外键相关的表创建管道并执行复制活动？

【英文标题】：How to create a pipeline and perform copy activity in Azure data factory for the tables related to foreign keys?

【发布时间】：2021-11-29 00:14:15

【问题描述】：

我是 Azure 数据工厂的新手，希望将 CSV 数据复制到具有外键关系的表中。这是我的表：

客户表

CREATE TABLE [dbo].[Customer]
(
    [Id] UNIQUEIDENTIFIER NOT NULL PRIMARY KEY, -- Primary Key column
    [CustomerNumber] NVARCHAR(50) NOT NULL,
    [FirstName] NVARCHAR(50) NOT NULL,
    [LastName] NVARCHAR(50) NOT NULL,
    [CreatedOn] datetime,
    [CreatedBy] NVARCHAR(255),
    [ModifiedOn] datetime,
    [ModifiedBy] NVARCHAR(255)
);
GO

-- Insert rows into table 'Customer' in schema '[dbo]'
INSERT INTO [dbo].[Customer]
VALUES
( 
 NEWID(),'Tom123', 'Tom', 'Shehu',GETDATE(),'test',GETDATE(),'admin'
),
( 
 NEWID(),'Harol234', 'Harold', 'Haoxa',GETDATE(),'test',GETDATE(),'admin'
),
( 
 NEWID(),'Peter345', 'Peter', 'Begu',GETDATE(),'test',GETDATE(),'admin'
),
( 
 NEWID(),'Marlin09', 'Marlin', 'Hysi',GETDATE(),'test',GETDATE(),'admin'
)

GO

产品表

CREATE TABLE [dbo].[Product]
(
    [Id] UNIQUEIDENTIFIER NOT NULL PRIMARY KEY, -- Primary Key column
    [Name] NVARCHAR(50) NOT NULL,
    [ErpNumber] NVARCHAR(50) NOT NULL,
    [Description] NVARCHAR(50) NOT NULL,
    [CreatedOn] datetime,
    [CreatedBy] NVARCHAR(255),
    [ModifiedOn] datetime,
    [ModifiedBy] NVARCHAR(255)
);
GO

-- Insert rows into table 'Product' in schema '[dbo]'
INSERT INTO [dbo].[Product]
VALUES
( 
 NEWID(), 'EI500CMZ', 'EI500CMZ','7-Day test product',GETDATE(),'Tom',GETDATE(),'Tom'
),
( 
 NEWID(), 'ST0SMX', 'ST0SMX','7-Day heavy duty product',GETDATE(),'Tom',GETDATE(),'Tom'
),
( 
 NEWID(), 'EH30MZ', 'EH30MZ','Electronic water test product',GETDATE(),'Tom',GETDATE(),'Tom'
)

CustomerProduct表

CREATE TABLE [dbo].[CustomerProduct]
(
    [Id] UNIQUEIDENTIFIER NOT NULL PRIMARY KEY, -- Primary Key column
    [CustomerId] UNIQUEIDENTIFIER NOT NULL,
    [ProductId] UNIQUEIDENTIFIER NOT NULL,
    [Name] NVARCHAR(255) NOT NULL,
    [CreatedOn] datetime,
    FOREIGN KEY(CustomerId) REFERENCES Customer(Id),
    FOREIGN KEY(ProductId) REFERENCES Product(Id)

);
GO

以下是我的 CSV 文件数据：CustomerNumber,ErpNumber,Name 汤姆123,EI500CMZ,EI500CMZ2340 Harol234,ST0SMX,ST0SMX74770 彼得345,EH30MZ,EH30MZ00234

现在我希望将数据插入到我的第三个表中，即 CustomerProduct，但我不明白如何插入“CustomerId”、“ProductId”和“Name”。 在上面的 CSV 数据中，我得到了“CustomerNumber”和“ErpNumber”，但在插入过程中，“CustomerId”和“ProductId”应该进入表格。

不明白怎么做。

到目前为止，我已经在 Azure 数据工厂中完成了这项工作：

创建了一个 Blob 存储帐户。在 Blob 存储中添加了一个容器并上传了我的 CSV 文件。

创建了一个名为“CustomerProductInputService”的 Azure blob 存储类型的链接服务，它将与 blob 存储通信

创建了一个名为“CustomerProductOutputService”的 Azure SQL 数据库类型的链接服务，该服务将与“CustomerProduct”表进行通信。

创建了一个 azure blob 类型的数据集。这将从“CustomerProductInputService”接收数据。

创建了一个 Azure SQL 数据库类型的数据集。

现在我被困在复制活动中。我不明白如何为此场景创建管道并将数据插入到 CustomerProduct 表中。 正如我解释的那样，我在 CSV 文件中获得了“CustomerNumber”和“ErpNumber”，但我想将“CustomerId”和“ProductId”插入到我的“CustomerProduct”表中。

谁能帮帮我？

【问题讨论】：

this article 是您要找的吗？

感谢此链接，我会检查链接。如果我用简单的 C# 语言说话，那么我会做这样的事情： foreach (DataRow item in csvData.Rows) string customerNumber = item["CustomerNumber"] .ToString(); Guid customerId = customerTable .Where(customer => customer.CustomerNumber == customerNumber) .Select(customer => customer.Id) .FirstOrDefault(); //我会做同样的事情来获取产品 ID

@ray 我的场景中需要数据流吗？

是的，我相信您的场景需要一个数据流。我认为仅使用动态内容来操作查找活动的输出不太可能实现您想要实现的按 customerNumber 行为连接。即使可行，我相信以后也很难维持动态内容。

【参考方案1】：

您可以使用 dataflow 活动将 CSV 中的 CustomerProduct 数据插入到表中，使用 lookup transformations 以获取 CustomerID em> 和 ProductID 分别来自 Customer 和 Product 表。

来源：

在数据流中添加 3 个源转换，1 个用于 CSV 源文件，1 个用于客户表，1 个用于产品表。

a) Source1 (CSV)：创建 CSV 数据集 到 source1 以获取 Input 文件数据。

b) Source2 (CustomerTable)：连接到 Customer 表，并从 Customer 表中获取所有现有数据。

• 由于我们只需要 Customer 表中的 ID 和 CustomerNumber 列，因此在 source2 之后添加 select 转换 (Customer) 以仅选择所需的列列表。

c) Source3 (ProductTable)：将 Source3 连接到 Product Table 以从 dbo.Product 中提取所有现有数据。

• 在 Source3 之后添加Select 转换（产品），以便从列列表中仅获取所需的列ID & ERPNumber。

将Lookup 转换添加到Source1 (CSV)，其中主流作为CSV 源和作为客户的查找流（源2 选择转换）和作为CSV 的查找条件“CustomerNumber”列等于 (==) 客户表列“CustomerNumber”。

由于Lookup 类似于此处的左连接，它在选择列表中包含来自 Source1 的所有列和来自 Source2 的 Lookup 列（包括重复列）。

a) 因此，使用select 转换 (CustomerSelectList) 仅选择输出中所需的列。还将从 Customer 表中提取的列名称“ID”重命名为 CustomerID，以与 Sink 表匹配。

在 Select (CustomerSelectList) 之后添加另一个 Lookup 转换以从 Product 表中获取数据。

a) 选择 Primary stream as CustomerSelectList（选择转换）和 Lookup stream 作为 Product（Select of Source3）

b) 查找条件为 CSV 源列“ErpNumber”等于 (==) 产品表列“ErpNumber”。

同样，使用select 转换忽略其他列并从选择列表中仅选择所需的列。还将列“ID”从 Product 表重命名为 ProductID。

将Derived Column 转换添加到选择(CustomerProductSelectList) 以添加新列ID 和CreatedOn。

a) ID：因为这是 sink 表中的 UNIQUEIDENTIFIER，我们可以添加一个表达式来使用 UUID() 生成 id在派生列中。

b) CreatedOn：添加表达式以获取 **Current timestamp** 到 Sink 表。

最后添加Sink 转换以将数据插入到CustomerProduct 表中。

将此数据流添加到管道并运行管道以插入数据。

输出：

【讨论】：

非常感谢。由于我是 ADF 的新手，这对我来说是一次很好的学习经历，它将帮助我解决问题。谢谢！！！

【参考方案2】：

首先，您需要确定客户与产品之间的关键联系。接下来，从数据工厂创建管道并创建 2 个源“Product”和“Customer”应用 ADF 转换 Join 和 Alter 并将其接收到 dbo.CustomerProduct。