使用 Azure 数据工厂或 Synapse Analytics 来复制和转换 Azure SQL 托管实例中的数据

适用于： Azure 数据工厂 Azure Synapse Analytics

本文概述如何“复制活动”功能从 Azure SQL 托管实例复制数据或将数据复制到其中，并使用“数据流”转换 Azure SQL 托管实例中的数据。 有关详细信息，请阅读 Azure 数据工厂和 Synapse Analytics 的简介文章。

支持的功能

此 Azure SQL 托管实例连接器支持以下功能：

① Azure 集成运行时 ② 自承载集成运行时

对于复制活动，此 Azure SQL 数据库连接器支持以下功能：

• 使用 SQL 身份验证和 Microsoft Entra 应用程序令牌身份验证通过服务主体或 Azure 资源托管标识来复制数据。
• 作为源，使用 SQL 查询或存储过程检索数据。 还可选择从 SQL MI 源进行并行复制。有关详细信息，请参阅从 SQL MI 进行并行复制部分。
• 作为接收器，根据源架构自动创建目标表（如果不存在）；在复制过程中，使用自定义逻辑将数据追加到表或调用存储过程。

先决条件

若要访问 SQL 托管实例公用终结点，可以使用托管的 Azure 集成运行时。 确保启用公共终结点，并在网络安全组中允许公共终结点流量，使该服务能够连接到你的数据库。 有关详细信息，请参阅此指南。

若要访问 SQL 托管实例专用终结点，请设置一个能够访问数据库的自承载集成运行时。 如果预配托管实例所在的虚拟网络中的自承载集成运行时，请确保集成运行时计算机与托管实例位于不同的子网中。 如果预配的自承载集成运行时与托管实例位于不同的虚拟网络中，则可使用虚拟网络对等互连或虚拟网络间的连接。 有关详细信息，请参阅将应用程序连接到 SQL 托管实例。

入门

若要使用管道执行复制活动，可以使用以下工具或 SDK 之一：

• 复制数据工具
• Azure 门户
• .NET SDK
• Python SDK
• Azure PowerShell
• REST API
• Azure 资源管理器模板

使用 UI 创建到 Azure SQL 托管实例的链接服务

使用以下步骤在 Azure 门户 UI 中创建到 SQL 托管实例的链接服务。

1. 浏览到 Azure 数据工厂或 Synapse 工作区中的“管理”选项卡并选择“链接服务”，然后单击“新建”：

   • Azure 数据工厂
   • Azure Synapse

   

2. 搜索“SQL”并选择 Azure SQL Server 托管实例连接器。

   

3. 配置服务详细信息、测试连接并创建新的链接服务。

   

连接器配置详细信息

以下部分详述了用来定义特定于 SQL 托管实例连接器的 Azure 数据工厂实体的属性。

链接服务属性

Azure SQL 托管实例连接器推荐版本支持 TLS 1.3。 请参阅此部分，了解如何将 Azure SQL 托管实例连接器版本从旧版升级。 有关属性详细信息，请参阅相应部分。

• 建议的版本
• 旧版本

建议的版本

应用推荐版本时，Azure SQL 托管实例链接服务支持以下通用属性：

有关其他连接属性，请参阅下表：

SQL 身份验证

若要使用 SQL 身份验证，除了前面部分所述的通用属性，还指定以下属性：

示例 1：使用 SQL 身份验证

{
    "name": "AzureSqlMILinkedService",
    "properties": {
        "type": "AzureSqlMI",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "SQL",
            "userName": "<user name>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

示例 2：将 SQL 身份验证与 Azure Key Vault 中的密码结合使用

{
    "name": "AzureSqlMILinkedService",
    "properties": {
        "type": "AzureSqlMI",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "SQL",
            "userName": "<user name>",
            "password": { 
                "type": "AzureKeyVaultSecret", 
                "store": { 
                    "referenceName": "<Azure Key Vault linked service name>", 
                    "type": "LinkedServiceReference" 
                }, 
                "secretName": "<secretName>" 
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

示例 3：将 SQL 身份验证与 Always Encrypted 配合使用

{
    "name": "AzureSqlMILinkedService",
    "properties": {
        "type": "AzureSqlMI",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "SQL",
            "userName": "<user name>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            }
        },
        "alwaysEncryptedSettings": {
            "alwaysEncryptedAkvAuthType": "ServicePrincipal",
            "servicePrincipalId": "<service principal id>",
            "servicePrincipalKey": {
                "type": "SecureString",
                "value": "<service principal key>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

服务主体身份验证

要使用服务主体身份验证，除了上一节中描述的通用属性外，还要指定以下属性

还需要执行以下步骤：

1. 按照步骤为托管实例预配 Microsoft Entra 管理员。

2. 在 Azure 门户中创建 Microsoft Entra 应用程序。 记下应用程序名称，以及以下定义链接服务的值：

   • 应用程序 ID
   • 应用程序密钥
   • 租户 ID

3. 为服务主体创建登录名。 在 SQL Server Management Studio (SSMS) 中，使用作为 sysadmin 的 SQL Server 帐户连接到托管实例。 在 master 数据库中，运行以下 T-SQL：

   CREATE LOGIN [your application name] FROM EXTERNAL PROVIDER
   

4. 为服务主体创建包含的数据库用户。 连接到要从其复制数据或将数据复制到其中的数据库，运行以下 T-SQL：

   CREATE USER [your application name] FROM EXTERNAL PROVIDER
   

5. 像通常对 SQL 用户和其他用户所做的那样向服务主体授予所需的权限。 运行以下代码。 有关更多选项，请参阅此文档。

   ALTER ROLE [role name e.g. db_owner] ADD MEMBER [your application name]
   

6. 配置 SQL 托管实例链接服务。

示例：使用服务主体身份验证

{
    "name": "AzureSqlDbLinkedService",
    "properties": {
        "type": "AzureSqlMI",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "hostNameInCertificate": "<host name>",
            "authenticationType": "ServicePrincipal",
            "servicePrincipalId": "<service principal id>",
            "servicePrincipalCredential": {
                "type": "SecureString",
                "value": "<application key>"
            },
            "tenant": "<tenant info, e.g. microsoft.partner.onmschina.cn>"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

系统分配的托管标识身份验证

数据工厂或 Synapse 工作区可以与系统分配的 Azure 资源的托管标识相关联，该托管标识代表了向其他 Azure 服务进行身份验证的服务。 可将此托管标识用于 SQL 托管实例身份验证。 指定的服务可以使用此标识访问数据库数据或从/向数据库复制数据。

若要使用系统分配的托管标识身份验证，请指定上一部分所述的通用属性，然后按照以下步骤操作。

1. 按照步骤为托管实例预配 Microsoft Entra 管理员。

2. 为系统分配的托管标识创建登录名。 在 SQL Server Management Studio (SSMS) 中，使用作为 sysadmin 的 SQL Server 帐户连接到托管实例。 在 master 数据库中，运行以下 T-SQL：

   CREATE LOGIN [your_factory_or_workspace_ name] FROM EXTERNAL PROVIDER
   

3. 为系统分配的托管标识创建包含的数据库用户。 连接到要从其复制数据或将数据复制到其中的数据库，运行以下 T-SQL：

   CREATE USER [your_factory_or_workspace_name] FROM EXTERNAL PROVIDER
   

4. 为系统分配的托管标识授予所需的权限，就像通常为 SQL 用户和其他用户所做的那样。 运行以下代码。 有关更多选项，请参阅此文档。

   ALTER ROLE [role name e.g. db_owner] ADD MEMBER [your_factory_or_workspace_name]
   

5. 配置 SQL 托管实例链接服务。

示例：使用系统分配的托管标识身份验证

{
    "name": "AzureSqlDbLinkedService",
    "properties": {
        "type": "AzureSqlMI",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "SystemAssignedManagedIdentity"
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

用户分配的托管标识身份验证

数据工厂或 Synapse 工作区可以与用户分配的托管标识相关联，该托管标识代表了向其他 Azure 服务进行身份验证的服务。 可将此托管标识用于 SQL 托管实例身份验证。 指定的服务可以使用此标识访问数据库数据或从/向数据库复制数据。

要使用用户分配的托管身份身份验证，除了上一节中描述的通用属性外，还要指定以下属性：

还需要执行以下步骤：

1. 按照步骤为托管实例预配 Microsoft Entra 管理员。

2. 为用户分配的托管标识创建登录名。 在 SQL Server Management Studio (SSMS) 中，使用作为 sysadmin 的 SQL Server 帐户连接到托管实例。 在 master 数据库中，运行以下 T-SQL：

   CREATE LOGIN [your_factory_or_workspace_ name] FROM EXTERNAL PROVIDER
   

3. 为用户分配的托管标识创建包含的数据库用户。 连接到要从其复制数据或将数据复制到其中的数据库，运行以下 T-SQL：

   CREATE USER [your_factory_or_workspace_name] FROM EXTERNAL PROVIDER
   

4. 创建一个或多个由用户分配的托管标识，并向用户分配的托管标识授予对 SQL 用户和其他用户正常操作所需的权限。 运行以下代码。 有关更多选项，请参阅此文档。

   ALTER ROLE [role name e.g. db_owner] ADD MEMBER [your_factory_or_workspace_name]
   

5. 为数据工厂分配一个或多个用户分配的托管标识，并为每个用户分配的托管标识创建凭据。

6. 配置 SQL 托管实例链接服务。

示例：使用用户分配的托管标识身份验证

{
    "name": "AzureSqlDbLinkedService",
    "properties": {
        "type": "AzureSqlMI",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "UserAssignedManagedIdentity",
            "credential": {
                "referenceName": "credential1",
                "type": "CredentialReference"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

旧版本

应用旧版本时，Azure SQL 托管实例链接服务支持以下通用属性：

有关各种身份验证类型，请参阅关于特定属性和先决条件的以下各部分：

• 旧版的 SQL 身份验证
• 旧版的服务主体身份验证
• 旧版本的系统分配的托管标识身份验证
• 旧版本的用户分配的托管标识身份验证

旧版的 SQL 身份验证

要使用 SQL 身份验证，请指定前面部分所述的通用属性。

旧版的服务主体身份验证

要使用服务主体身份验证，除了上一部分中描述的通用属性外，还需要指定以下属性：

还需要按照服务主体身份验证中的步骤授予相应的权限。

旧版本的系统分配的托管标识身份验证

若要使用系统分配的托管标识身份验证，请按照系统分配的托管标识身份验证，为推荐的版本执行相同的步骤。

旧版本的用户分配的托管标识身份验证

若要使用用户分配的托管标识身份验证，请按照用户分配的托管标识身份验证，为推荐的版本执行相同的步骤。

数据集属性

如需可用于定义数据集的节和属性的完整列表，请参阅有关数据集的文章。 此部分提供 SQL 托管实例数据集支持的属性列表。

若要从/向 SQL 托管实例复制数据，以下属性需受支持：

示例

{
    "name": "AzureSqlMIDataset",
    "properties":
    {
        "type": "AzureSqlMITable",
        "linkedServiceName": {
            "referenceName": "<SQL Managed Instance linked service name>",
            "type": "LinkedServiceReference"
        },
        "schema": [ < physical schema, optional, retrievable during authoring > ],
        "typeProperties": {
            "schema": "<schema_name>",
            "table": "<table_name>"
        }
    }
}

复制活动属性

有关可用于定义活动的节和属性的完整列表，请参阅管道一文。 此部分提供 SQL 托管实例源和和接收器支持的属性列表。

以 SQL 托管实例作为源

若要从 SQL 托管实例复制数据，复制活动的 source 节需要支持以下属性：

请注意以下几点：

• 如果为 SqlMISource 指定 sqlReaderQuery，则复制活动会针对 SQL 托管实例源运行此查询以获取数据。 也可通过指定 sqlReaderStoredProcedureName 和 storedProcedureParameters 来指定存储过程，前提是存储过程使用参数 。
• 在源中使用存储过程检索数据时，请注意，如果存储过程设计为当传入不同的参数值时返回不同的架构，则从 UI 导入架构或使用自动表创建将数据复制到 SQL 数据库时，可能会遇到故障或出现意外的结果。

示例：使用 SQL 查询

"activities":[
    {
        "name": "CopyFromAzureSqlMI",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<SQL Managed Instance input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "SqlMISource",
                "sqlReaderQuery": "SELECT * FROM MyTable"
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

示例：使用存储过程

"activities":[
    {
        "name": "CopyFromAzureSqlMI",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<SQL Managed Instance input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "SqlMISource",
                "sqlReaderStoredProcedureName": "CopyTestSrcStoredProcedureWithParameters",
                "storedProcedureParameters": {
                    "stringData": { "value": "str3" },
                    "identifier": { "value": "$$Text.Format('{0:yyyy}', <datetime parameter>)", "type": "Int"}
                }
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

存储过程定义

CREATE PROCEDURE CopyTestSrcStoredProcedureWithParameters
(
    @stringData varchar(20),
    @identifier int
)
AS
SET NOCOUNT ON;
BEGIN
    select *
    from dbo.UnitTestSrcTable
    where dbo.UnitTestSrcTable.stringData != stringData
    and dbo.UnitTestSrcTable.identifier != identifier
END
GO

以 SQL 托管实例作为接收器

若要将数据复制到 SQL 托管实例，复制活动的 sink 节需要支持以下属性：

示例 1：追加数据

"activities":[
    {
        "name": "CopyToAzureSqlMI",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<SQL Managed Instance output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "SqlMISink",
                "tableOption": "autoCreate",
                "writeBatchSize": 100000
            }
        }
    }
]

示例 2：在复制过程中调用存储过程

请参阅调用 SQL MI 接收器的存储过程，了解更多详细信息。

"activities":[
    {
        "name": "CopyToAzureSqlMI",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<SQL Managed Instance output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "SqlMISink",
                "sqlWriterStoredProcedureName": "CopyTestStoredProcedureWithParameters",
                "storedProcedureTableTypeParameterName": "MyTable",
                "sqlWriterTableType": "MyTableType",
                "storedProcedureParameters": {
                    "identifier": { "value": "1", "type": "Int" },
                    "stringData": { "value": "str1" }
                }
            }
        }
    }
]

示例 3：更新插入数据

"activities":[
    {
        "name": "CopyToAzureSqlMI",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<SQL Managed Instance output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "SqlMISink",
                "tableOption": "autoCreate",
                "writeBehavior": "upsert",
                "upsertSettings": {
                    "useTempDB": true,
                    "keys": [
                        "<column name>"
                    ]
                },            
            }
        }
    }
]

从 SQL MI 进行并行复制

复制活动中的 Azure SQL 托管实例连接器提供内置的数据分区，用于并行复制数据。 可以在复制活动的“源”表中找到数据分区选项。

启用分区复制时，复制活动将对 SQL MI 源运行并行查询，以按分区加载数据。 可通过复制活动中的 parallelCopies 设置控制并行度。 例如，如果将 parallelCopies 设置为 4，则该服务会根据指定的分区选项和设置并行生成并运行 4 个查询，每个查询从 SQL MI 检索一部分数据。

建议同时启用并行复制和数据分区，尤其是从 SQL MI 加载大量数据时。 下面是适用于不同方案的建议配置。 将数据复制到基于文件的数据存储中时，建议将数据作为多个文件写入文件夹（仅指定文件夹名称），在这种情况下，性能优于写入单个文件。

使用分区选项加载数据的最佳做法：

1. 选择独特的列作为分区列（如主键或唯一键），以避免数据倾斜。
2. 如果表具有内置分区，请使用名为“表的物理分区”分区选项来提升性能。
3. 如果使用 Azure Integration Runtime 复制数据，则可设置较大的“数据集成单元 (DIU)”(>4) 以利用更多计算资源。 检查此处适用的方案。
4. “复制并行度”可控制分区数量，将此数字设置得太大有时会损害性能，建议将此数字设置按以下公式计算的值：（DIU 或自承载 IR 节点数）*（2 到 4）。

示例：从包含物理分区的大型表进行完整加载

"source": {
    "type": "SqlMISource",
    "partitionOption": "PhysicalPartitionsOfTable"
}

示例：使用动态范围分区进行查询

"source": {
    "type": "SqlMISource",
    "query": "SELECT * FROM <TableName> WHERE ?DfDynamicRangePartitionCondition AND <your_additional_where_clause>",
    "partitionOption": "DynamicRange",
    "partitionSettings": {
        "partitionColumnName": "<partition_column_name>",
        "partitionUpperBound": "<upper_value_of_partition_column (optional) to decide the partition stride, not as data filter>",
        "partitionLowerBound": "<lower_value_of_partition_column (optional) to decide the partition stride, not as data filter>"
    }
}

检查物理分区的示例查询

SELECT DISTINCT s.name AS SchemaName, t.name AS TableName, pf.name AS PartitionFunctionName, c.name AS ColumnName, iif(pf.name is null, 'no', 'yes') AS HasPartition
FROM sys.tables AS t
LEFT JOIN sys.objects AS o ON t.object_id = o.object_id
LEFT JOIN sys.schemas AS s ON o.schema_id = s.schema_id
LEFT JOIN sys.indexes AS i ON t.object_id = i.object_id 
LEFT JOIN sys.index_columns AS ic ON ic.partition_ordinal > 0 AND ic.index_id = i.index_id AND ic.object_id = t.object_id 
LEFT JOIN sys.columns AS c ON c.object_id = ic.object_id AND c.column_id = ic.column_id 
LEFT JOIN sys.partition_schemes ps ON i.data_space_id = ps.data_space_id 
LEFT JOIN sys.partition_functions pf ON pf.function_id = ps.function_id 
WHERE s.name='[your schema]' AND t.name = '[your table name]'

如果表具有物理分区，你可看到“HasPartition”为“是”，如下所示。

将数据加载到 SQL 托管实例中的最佳做法

将数据复制到 SQL 托管实例中时，可能需要不同的写入行为：

• 追加：我的源数据只包含新记录。
• 更新插入：我的源数据包含插入和更新内容。
• 覆盖：我需要每次都重新加载整个维度表。
• 使用自定义逻辑进行写入：在将数据最终插入目标表之前，我需要额外的处理。

有关配置步骤和最佳做法，请参阅相应的部分。

追加数据

追加数据是此 SQL 托管实例接收器连接器的默认行为。 该服务执行批量插入，以有效地在表中写入数据。 可以相应地在复制活动中配置源和接收器。

更新插入数据

复制活动现在支持以本机方式将数据加载到数据库临时表中，并在键存在时更新接收器表中的数据，不存在时插入新数据。 若要详细了解复制活动中的更新插入设置，请参阅 以 SQL 托管实例作为接收器。

覆盖整个表

可以在复制活动接收器中配置 preCopyScript 属性。 在这种情况下，对于运行的每个复制活动，该服务首先运行脚本。 然后，运行复制来插入数据。 例如，若要使用最新数据覆盖整个表，请指定一个脚本，以先删除所有记录，然后从源批量加载新数据。

使用自定义逻辑写入数据

使用自定义逻辑写入数据的步骤与更新插入数据部分中的描述类似。 如果在将源数据最终插入目标表之前需要应用额外的处理，则可先将数据加载到临时表，然后再调用存储过程活动，或者在复制活动接收器中调用存储过程来应用数据。

调用 SQL 接收器的存储过程

将数据复制到 SQL 托管实例中时，还可以通过对每批的源表使用更多参数来配置和调用用户指定的存储过程。 存储过程功能利用表值参数。

当内置复制机制无法使用时，还可使用存储过程。 例如，在将源数据最终插入目标表之前应用额外的处理。 额外处理的示例包括合并列、查找其他值以及将数据插入多个表。

以下示例演示如何使用存储过程，在 SQL Server 数据库中的表内执行 upsert。 假设输入数据和接收器 Marketing 表各有三列：ProfileID、State 和 Category。 基于 ProfileID 列执行更新插入，并仅将其应用于名为“ProductA”的特定类别。

1. 在数据库中，使用与 sqlWriterTableType 相同的名称定义表类型。 表类型的架构与输入数据返回的架构相同。

   CREATE TYPE [dbo].[MarketingType] AS TABLE(
       [ProfileID] [varchar](256) NOT NULL,
       [State] [varchar](256) NOT NULL,
       [Category] [varchar](256) NOT NULL
   )
   

2. 在数据库中，使用与 sqlWriterStoredProcedureName 相同的名称定义存储过程。 它可处理来自指定源的输入数据，并将其合并到输出表中。 存储过程中的表类型的参数名称与数据集中定义的 tableName 相同。

   CREATE PROCEDURE spOverwriteMarketing @Marketing [dbo].[MarketingType] READONLY, @category varchar(256)
   AS
   BEGIN
   MERGE [dbo].[Marketing] AS target
   USING @Marketing AS source
   ON (target.ProfileID = source.ProfileID and target.Category = @category)
   WHEN MATCHED THEN
       UPDATE SET State = source.State
   WHEN NOT MATCHED THEN
       INSERT (ProfileID, State, Category)
       VALUES (source.ProfileID, source.State, source.Category);
   END
   

3. 在管道中，在复制活动中定义 SQL MI 接收器部分，如下所示：

   "sink": {
       "type": "SqlMISink",
       "sqlWriterStoredProcedureName": "spOverwriteMarketing",
       "storedProcedureTableTypeParameterName": "Marketing",
       "sqlWriterTableType": "MarketingType",
       "storedProcedureParameters": {
           "category": {
               "value": "ProductA"
           }
       }
   }
   

映射数据流属性

在映射数据流中转换数据时，可以在 Azure SQL 托管实例中读取表以及将数据写入表。 有关详细信息，请参阅映射数据流中的源转换和接收器转换。

源转换

下表列出了 Azure SQL 托管实例源支持的属性。 你可以在“源选项”选项卡中编辑这些属性。

CREATE PROC CTESP @query nvarchar(max)
AS
BEGIN
EXECUTE sp_executesql @query;
END

Azure SQL 托管实例源脚本示例

使用 Azure SQL 托管实例作为源类型时，关联的数据流脚本为：

source(allowSchemaDrift: true,
    validateSchema: false,
    isolationLevel: 'READ_UNCOMMITTED',
    query: 'select * from MYTABLE',
    format: 'query') ~> SQLMISource

接收器转换

下表列出了 Azure SQL 托管实例接收器支持的属性。 可以在“接收器选项”选项卡中编辑这些属性。

Azure SQL 托管实例接收器脚本示例

使用 Azure SQL 托管实例作为接收器类型时，关联的数据流脚本为：

IncomingStream sink(allowSchemaDrift: true,
    validateSchema: false,
    deletable:false,
    insertable:true,
    updateable:true,
    upsertable:true,
    keys:['keyColumn'],
    format: 'table',
    skipDuplicateMapInputs: true,
    skipDuplicateMapOutputs: true) ~> SQLMISink

Lookup 活动属性

若要了解有关属性的详细信息，请查看 Lookup 活动。

GetMetadata 活动属性

若要了解有关属性的详细信息，请查看 GetMetadata 活动

SQL 托管实例的数据类型映射

使用复制活动向/从 SQL 托管实例复制数据时，会使用以下从 SQL 托管实例数据类型到服务内部使用的临时数据类型映射。 若要了解复制活动如何从源架构和数据类型映射到接收器，请参阅架构和数据类型映射。

使用 Always Encrypted

使用 Always Encrypted 从/向 SQL 托管实例复制数据时，请执行以下步骤：

1. 将列主密钥 (CMK) 存储在 Azure 密钥保管库中。 详细了解如何使用 Azure 密钥保管库配置 Always Encrypted

2. 确保授予对存储了列主密钥 (CMK) 的密钥保管库的访问权限。 有关所需的权限，请参阅此文。

3. 创建链接服务，以使用托管标识或服务主体连接到 SQL 数据库并启用“Always Encrypted”功能。

本机变更数据捕获

Azure 数据工厂可针对 SQL Server、Azure SQL DB 和 Azure SQL MI 支持本机变更数据捕获功能。 可以通过 ADF 映射数据流自动检测和提取 SQL 存储中的行插入、更新和删除等变更数据。 由于映射数据流没有代码体验，用户可以通过将数据库追加为目标存储来轻松实现 SQL 存储中的数据复制方案。 更重要的是，用户还可以在两者之间撰写任何数据转换逻辑，以便从 SQL 存储实现增量 ETL 方案。

请确保管道和活动名称保持不变，以便 ADF 可以记录检查点，从而自动从上次运行中获取更改的数据。 如果更改管道名称或活动名称，检查点将重置，进而导致从头开始或是在下一次运行中开始获取现在开始的更改数据。 如果要更改管道名称或活动名称，但仍保留检查点，以自动从上次运行中获取变更的数据，请使用自己的数据流活动中的检查点密钥来实现此目的。

调试管道时，此功能都以相同的方式工作。 请注意，在调试运行期间刷新浏览器时，检查点将重置。 若对调试运行中的管道结果感到满意，可继续发布并触发管道。 首次触发已发布管道时，管道将自动从头开始重启，或者立即开始获取更改数据。

在监视部分，你始终有机会重新运行管道。 执行此操作时，始终可从所选管道运行的上一个检查点捕获已更改的数据。

示例 1：

将引用到已启用 SQL CDC 的数据集的源转换与映射数据流中引用到数据库的接收器转换直接链接起来时，SQL 源上发生的变更将自动应用到目标数据库，这样很轻松就可以获得数据库之间的数据复制方案。 可以使用接收器转换中的更新方法选择是否在目标数据库上允许插入、允许更新或允许删除。 映射数据流中的示例脚本如下所示。

source(output(
		id as integer,
		name as string
	),
	allowSchemaDrift: true,
	validateSchema: false,
	enableNativeCdc: true,
	netChanges: true,
	skipInitialLoad: false,
	isolationLevel: 'READ_UNCOMMITTED',
	format: 'table') ~> source1
source1 sink(allowSchemaDrift: true,
	validateSchema: false,
	deletable:true,
	insertable:true,
	updateable:true,
	upsertable:true,
	keys:['id'],
	format: 'table',
	skipDuplicateMapInputs: true,
	skipDuplicateMapOutputs: true,
	errorHandlingOption: 'stopOnFirstError') ~> sink1

示例 2：

如果想启用 ETL 方案，而不是通过 SQL CDC 在数据库之间复制数据，则可以使用映射数据流中的表达式，包括 isInsert(1)、isUpdate(1) 和 isDelete(1) 来区分具有不同操作类型的行。 下面是映射数据流的示例脚本之一，它派生一个列以及相关值：1 表示插入的行，2 表示更新的行，3 表示删除的行，用于下游转换以处理增量数据。

source(output(
		id as integer,
		name as string
	),
	allowSchemaDrift: true,
	validateSchema: false,
	enableNativeCdc: true,
	netChanges: true,
	skipInitialLoad: false,
	isolationLevel: 'READ_UNCOMMITTED',
	format: 'table') ~> source1
source1 derive(operationType = iif(isInsert(1), 1, iif(isUpdate(1), 2, 3))) ~> derivedColumn1
derivedColumn1 sink(allowSchemaDrift: true,
	validateSchema: false,
	skipDuplicateMapInputs: true,
	skipDuplicateMapOutputs: true) ~> sink1

已知限制：

• 只有来自 SQL CDC 的净变更将通过 cdc.fn_cdc_get_net_changes_ 由 ADF 加载。

升级 Azure SQL 托管实例版本

要升级 Azure SQL 托管实例版本，请在“编辑链接服务”页中，选择“版本”下的“推荐”，并参考推荐版本的链接服务属性来配置链接服务。

建议版本和旧版本之间的差异

下表显示了使用建议版本和旧版本的 Azure SQL 托管实例之间的功能差异。

相关内容

有关复制活动支持作为源和接收器的数据存储的列表，请参阅受支持的数据存储。