通过使用 Azure 数据工厂或 Azure Synapse Analytics 从/向 SQL Server 复制和转换数据

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

本文概述了如何使用 Azure 数据工厂和 Azure Synapse 管道中的复制活动从/向 SQL Server 数据库复制数据，以及如何使用数据流来转换 SQL Server 数据库中的数据。 有关详细信息，请阅读 Azure 数据工厂或 Azure Synapse Analytics 的简介文章。

支持的功能

此 SQL Server 连接器支持以下功能：

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

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

具体而言，此 SQL Server 连接器支持：

• SQL Server 版本 2005 及更高版本。
• 使用 SQL 或 Windows 身份验证复制数据。
• 作为源，使用 SQL 查询或存储过程检索数据。 你还可以选择从 SQL Server 源进行并行复制，请参阅从 SQL 数据库进行并行复制部分获取详细信息。
• 作为接收器，根据源架构自动创建目标表（如果不存在）；在复制过程中，将数据追加到表或使用自定义逻辑调用存储过程。

SQL Server Express LocalDB 不受支持。

先决条件

如果数据存储位于本地网络、Azure 虚拟网络或 Amazon Virtual Private Cloud 内部，则需要配置自承载集成运行时才能连接到该数据存储。

如果数据存储是托管的云数据服务，则可以使用 Azure Integration Runtime。 如果访问范围限制为防火墙规则中允许的 IP，你可以选择将 Azure Integration Runtime IP 添加到允许列表。

此外，还可以使用 Azure 数据工厂中的托管虚拟网络集成运行时功能访问本地网络，而无需安装和配置自承载集成运行时。

要详细了解网络安全机制和数据工厂支持的选项，请参阅数据访问策略。

入门

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

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

使用 UI 创建 SQL Server 链接服务

使用以下步骤在 Azure 门户 UI 中创建 SQL Server 链接服务。

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

   • Azure 数据工厂
   • Azure Synapse

   

2. 搜索“SQL”并选择“SQL Server 连接器”。

   

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

   

连接器配置详细信息

对于特定于 SQL Server 数据库连接器的数据工厂和 Synapse 管道实体，以下部分提供有关用于定义这些实体的属性的详细信息。

链接服务属性

SQL Server 推荐版本支持 TLS 1.3。 如果你使用旧版本，请参阅此部分升级你的 SQL Server 链接服务。 有关属性详细信息，请参阅相应部分。

• 建议的版本
• 旧版本

建议的版本

应用推荐版本时，SQL Server 链接服务支持以下通用属性：

有关其他连接属性，请查看下表：

SQL 身份验证

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

示例：使用 SQL 身份验证

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "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"
        }
    }
}

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

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "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"
        }
    }
}

示例：使用 Always Encrypted

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "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"
        }
    }
}

Windows 身份验证

若要使用 Windows 身份验证，则除了上一部分所述的通用属性外，还需要指定以下属性：

示例：使用 Windows 身份验证

{
    "name": "SqlServerLinkedService",
    "properties": {
        "type": "SqlServer",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "Windows",
            "userName": "<domain\\username>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

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

{
    "name": "SqlServerLinkedService",
    "properties": {
        "annotations": [],
        "type": "SqlServer",
        "typeProperties": {
            "server": "<name or network address of the SQL server instance>",
            "database": "<database name>",
            "encrypt": "<encrypt>",
            "trustServerCertificate": false,
            "authenticationType": "Windows",
            "userName": "<domain\\username>",
            "password": {
                "type": "AzureKeyVaultSecret",
                "store": {
                    "referenceName": "<Azure Key Vault linked service name>",
                    "type": "LinkedServiceReference"
                },
                "secretName": "<secretName>"
            }
        },
        "connectVia": {
            "referenceName": "<name of Integration Runtime>",
            "type": "IntegrationRuntimeReference"
        }
    }
}

旧版本

应用旧版本时，SQL Server 链接服务支持以下通用属性：

此 SQL Server 连接器支持以下身份验证类型。 有关详细信息，请参阅相应部分。

• 旧版本的 SQL 身份验证
• 旧版本的 Windows 身份验证

旧版本的 SQL 身份验证

若要使用 SQL 身份验证，则除了上一部分所述的通用属性外，还需要指定以下属性：

旧版本的 Windows 身份验证

若要使用 Windows 身份验证，则除了上一部分所述的通用属性外，还需要指定以下属性：

数据集属性

有关可用于定义数据集的各部分和属性的完整列表，请参阅数据集一文。 本部分提供 SQL Server 数据集支持的属性列表。

从/向 SQL Server 数据库复制数据时支持以下属性：

示例

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

复制活动属性

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

SQL Server 作为源

要从 SQL Server 复制数据，请将复制活动中的源类型设置为 SqlSource。 复制活动的 source 节支持以下属性：

请注意以下几点：

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

示例：使用 SQL 查询

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

示例：使用存储过程

"activities":[
    {
        "name": "CopyFromSQLServer",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<SQL Server input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "SqlSource",
                "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 Server 作为接收器

要向 SQL Server 复制数据，请将复制活动中的接收器类型设置为 SqlSink。 复制活动的 sink 节支持以下属性：

示例 1：追加数据

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

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

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

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

示例 3：更新插入数据

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

从 SQL 数据库进行并行复制

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

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

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

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

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

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

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

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

"source": {
    "type": "SqlSource",
    "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 Server 中的最佳做法

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

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

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

追加数据

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

更新插入数据

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

覆盖整个表

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

使用自定义逻辑写入数据

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

调用 SQL 接收器的存储过程

将数据复制到 SQL Server 数据库中时，还可以通过对每批源表使用附加参数来配置和调用用户指定的存储过程。 存储过程功能利用表值参数。 请注意，该服务会自动将存储过程包装到其自己的事务中，因此在存储过程内创建的任何事务都将成为嵌套事务，并且可能会影响异常处理。

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

以下示例演示如何使用存储过程，在 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 sink 节。

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

映射数据流属性

在转换映射数据流中的数据时，可以从 SQL Server 数据库读取表和写入表。 有关详细信息，请参阅映射数据流中的源转换和接收器转换。

源转换

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

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

SQL Server 源脚本示例

在使用 SQL Server 作为源类型时，关联的数据流脚本为：

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

接收器转换

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

SQL Server 接收器脚本示例

在使用 SQL Server 作为接收器类型时，关联的数据流脚本为：

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

SQL Server 的数据类型映射

从/向 SQL Server 复制数据时，以下映射用于从 SQL Server 数据类型映射到 Azure 数据工厂临时数据类型。 实现数据工厂的 Synapse 管道使用相同的映射。 若要了解复制活动如何将源架构和数据类型映射到接收器，请参阅架构和数据类型映射。

查找活动属性

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

GetMetadata 活动属性

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

使用 Always Encrypted

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

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 加载。

排查连接问题

1. 将 SQL Server 实例配置为接受远程连接。 启动“SQL Server Management Studio”，右键单击“服务器”并选择“属性” 。 从列表中选择“连接”，并选中“允许远程连接到此服务器”复选框 。

   

   有关详细步骤，请参阅配置远程访问服务器配置选项。

2. 启动“SQL Server 配置管理器”。 针对所需实例展开“SQL Server 网络配置”，并选择“MSSQLSERVER 的协议” 。 协议将显示在右窗格中。 右键单击“TCP/IP”并选择“启用”以启用 TCP/IP 。

   

   有关详细信息和启用 TCP/IP 协议的其他方法，请参阅启用或禁用服务器网络协议。

3. 在同一窗口中，双击“TCP/IP”以启动“TCP/IP 属性”窗口 。

4. 切换到“IP 地址”选项卡。向下滚动到“IPAll”部分。 记下“TCP 端口”的值。 默认值为 1433。

5. 在计算机上创建 Windows 防火墙规则，以便允许通过此端口传入流量。

6. 验证连接：若要使用完全限定名称连接到 SQL Server，请从另一台计算机使用 SQL Server Management Studio。 示例为 "<machine>.<domain>.corp.<company>.com,1433"。

升级 SQL Server 版本

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

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

下表显示了使用建议版本和旧版本的 SQL Server 之间的差异。

相关内容

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