azure - 在Azure数据工厂中，是否可以使用另一个源中的列来过滤复制事件的源查询？

Question

我有一个发票表，11M 行。我想停止每天进行完全刷新并增量加载。我本质上是想更新更改的行并添加新行。

我的想法是创建一个表键并使用它来确定发票行是否已存在于我的接收器中。

案例:源 key 是...

• 在水槽中:什么都不做
• 不在接收器中:添加到接收器

案例:水槽 key 是...

• 不在来源中:删除

想象我要做什么......

但是，我的问题是我的 Source 和 Sink 不是同一类型。源 = ODBC (Hive)，接收器 = Azure SQL 数据库。因此，在 ADF 中，您无法加入它们，并且查找的行数限制为 5000 行。

有什么办法可以解决这个问题吗？无论如何要在另一个的Where 子句中使用一个的列吗？

Answer 1

我们可以使用MERGE为了实现这一点。合并语法如下:

MERGE target_table USING source_table
ON merge_condition
WHEN MATCHED
    THEN update_statement
WHEN NOT MATCHED
    THEN insert_statement
WHEN NOT MATCHED BY SOURCE

1. 我还创建了一个简单的测试。我的源是 Azure 数据湖中的 emp.csv。接收器是我的 Azure SQL。我创建一个表和一个表类型。

CREATE TABLE [dbo].[emp](
    [id] [int] NOT NULL,
    [name] [varchar](20) NULL,
    [age] [int] NULL
)

CREATE TYPE [dbo].[EmpType] AS TABLE(
    [id] [int] NOT NULL,
    [name] [nvarchar](max) NOT NULL,
    [age] [nvarchar](max) NOT NULL
)
GO

然后，我创建存储过程来更新和插入行。这里有一个小问题，azure sql似乎不支持更新操作后的删除操作。所以我定义了另一个专门用于删除操作的存储过程。

CREATE PROCEDURE [dbo].[uspEmp]

@emp [dbo].[EmpType] READONLY

AS
        MERGE [dbo].[emp] AS target_sqldb

        USING @emp AS source_tblstg

        ON target_sqldb.id = source_tblstg.id 

        WHEN MATCHED THEN

        UPDATE SET

        target_sqldb.name = source_tblstg.name,

        target_sqldb.age = source_tblstg.age
        
        WHEN NOT MATCHED BY TARGET THEN 

        INSERT VALUES (

            source_tblstg.id,

            source_tblstg.name,

            source_tblstg.age
        );

        --WHEN NOT MATCHED BY SOURCE THEN
        --DELETE;

在 ADF 中，我们可以使用Copy Activity1 将行复制到 Azure SQL 中。选择存储的过程，然后点击导入参数。

然后我们可以使用 Coy Activity2 从 Azure SQL 中删除行。

CREATE PROCEDURE [dbo].[deleteEmp]

@emp [dbo].[EmpType] READONLY

AS
        MERGE [dbo].[emp] AS target_sqldb

        USING @emp AS source_tblstg

        ON target_sqldb.id = source_tblstg.id 
        
        WHEN NOT MATCHED BY SOURCE THEN
        DELETE;

同样，选择存储的程序，然后点击导入参数。

我的测试成功，没有删除源中的行，而是更新并插入了源中的行。