使用 Azure 数据工厂或 Synapse Analytics 来复制和转换 Azure Database for PostgreSQL 中的数据

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

本文概述了如何使用 Azure 数据工厂和 Synapse Analytics 管道中的复制活动从/向 Azure Database for PostgreSQL 复制数据，以及如何使用数据流转换 Azure Database for PostgreSQL 中的数据。 若要了解详细信息，请阅读 Azure 数据工厂 和 Synapse Analytics 的介绍性文章。

此连接器专用于 Azure Database for PostgreSQL 服务。 若要从位于本地或云中的通用 PostgreSQL 数据库复制数据，请使用 PostgreSQL 连接器。

此 Azure Database for PostgreSQL 连接器支持以下功能：

（1） Azure 集成运行时 （2） 自承载集成运行时

这三个活动适用于 Azure Database for PostgreSQL 单一服务器 和 灵活服务器，以及 Azure Cosmos DB for PostgreSQL。

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

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

使用以下步骤在 Azure 门户 UI 中创建一个到 Azure Database for PostgreSQL 的链接服务。

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

   • Azure 数据工厂
   • Azure Synapse

   

2. 搜索 PostgreSQL 并选择 Azure Database for PostgreSQL 连接器。

   

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

   

以下部分提供详细信息，描述用于定义特定于 Azure Database for PostgreSQL 连接器的数据工厂实体的属性。

Azure Database for PostgreSQL 连接器版本 2.0 支持 TLS 1.3 和多个 SSL 模式。 请参阅本 部分 ，从版本 1.0 升级 Azure SQL 数据库连接器版本。 有关属性详细信息，请参阅相应部分。

• 版本 2.0
• 版本 1.0

应用版本 2.0 时，Azure Database for PostgreSQL 链接服务支持以下属性：

示例：

{
    "name": "AzurePostgreSqlLinkedService",
    "properties": {
        "type": "AzurePostgreSql",
        "version": "2.0",
        "typeProperties": {
            "server": "<server name>",
            "port": "5432",
            "database": "<database name>",
            "sslMode": 2,
            "username": "<user name>",
            "password": {
                "type": "SecureString",
                "value": "<password>"
            }
        }
    }
}

示例：

在 Azure Key Vault 中存储密码

{
    "name": "AzurePostgreSqlLinkedService",
    "properties": {
        "type": "AzurePostgreSql",
        "version": "2.0",
        "typeProperties": {
            "server": "<server name>",
            "port": "5432",
            "database": "<database name>",
            "sslMode": 2,
            "username": "<user name>",
            "password": { 
                "type": "AzureKeyVaultSecret", 
                "store": { 
                    "referenceName": "<Azure Key Vault linked service name>", 
                    "type": "LinkedServiceReference" 
                }, 
                "secretName": "<secretName>" 
            }
        }
    }
}

示例：

服务主体密钥

{
    "name": "AzurePostgreSqlLinkedService",
    "type": "Microsoft.DataFactory/factories/linkedservices",
    "properties": {
        "annotations": [],
        "type": "AzurePostgreSql",
        "version": "2.0",
        "typeProperties": {
            "server": "<server name>",
            "port": 5432,
            "database": "<database name>",
            "sslMode": 2,
            "username": "<service principal name>",
            "authenticationType": "<authentication type>",
            "tenant": "<tenant>",
            "servicePrincipalId": "<service principal ID>",
            "azureCloudType": "<azure cloud type>",
            "servicePrincipalCredentialType": "<service principal type>",
            "servicePrincipalKey": "<service principal key>"
        }
    }
}

示例：

服务主体证书

{
    "name": "AzurePostgreSqlLinkedService",
    "type": "Microsoft.DataFactory/factories/linkedservices",
    "properties": {
        "annotations": [],
        "type": "AzurePostgreSql",
        "version": "2.0",
        "typeProperties": {
            "server": "<server name>",
            "port": 5432,
            "database": "<database name>",
            "sslMode": 2,
            "username": "<service principal name>",
            "authenticationType": "<authentication type>",
            "tenant": "<tenant>",
            "servicePrincipalId": "<service principal ID>",
            "azureCloudType": "<azure cloud type>",
            "servicePrincipalCredentialType": "<service principal type>",
            "servicePrincipalEmbeddedCert": "<service principal certificate>",
            "servicePrincipalEmbeddedCertPassword": "<service principal embedded certificate password>"
        }
    }
}

应用版本 1.0 时，Azure Database for PostgreSQL 链接服务支持以下属性：

典型的连接字符串为 host=<server>.postgres.database.chinacloudapi.cn;database=<database>;port=<port>;uid=<username>;password=<password>。 以下是你可以根据具体情况设置的更多属性：

示例：

{
    "name": "AzurePostgreSqlLinkedService",
    "properties": {
        "type": "AzurePostgreSql",
        "version": "1.0",
        "typeProperties": {
            "connectionString": "host=<server>.postgres.database.chinacloudapi.cn;database=<database>;port=<port>;uid=<username>;password=<password>"
        }
    }
}

示例：

在 Azure Key Vault 中存储密码

{
    "name": "AzurePostgreSqlLinkedService",
    "properties": {
        "type": "AzurePostgreSql",
        "version": "1.0",
        "typeProperties": {
            "connectionString": "host=<server>.postgres.database.chinacloudapi.cn;database=<database>;port=<port>;uid=<username>;",
            "password": { 
                "type": "AzureKeyVaultSecret", 
                "store": { 
                    "referenceName": "<Azure Key Vault linked service name>", 
                    "type": "LinkedServiceReference" 
                }, 
                "secretName": "<secretName>" 
            }
        }
    }
}

有关可用于定义数据集的部分和属性的完整列表，请参阅 “数据集”。 本部分提供数据集中 Azure Database for PostgreSQL 支持的属性列表。

若要从 Azure Database for PostgreSQL 复制数据，请将数据集的 type 属性设置为 AzurePostgreSqlTable。 支持以下属性：

示例：

{
    "name": "AzurePostgreSqlDataset",
    "properties": {
        "type": "AzurePostgreSqlTable",
        "linkedServiceName": {
            "referenceName": "<AzurePostgreSql linked service name>",
            "type": "LinkedServiceReference"
        },
        "typeProperties": {
            "schema": "<schema_name>",
            "table": "<table_name>"
        }
    }
}

有关可用于定义活动的各个部分和属性的完整列表，请参阅管道和活动。 本部分提供 Azure Database for PostgreSQL 源支持的属性列表。

若要从 Azure Database for PostgreSQL 复制数据，请将复制活动中的源类型设置为 AzurePostgreSqlSource。 复制活动 源 部分支持以下属性：

示例：

"activities":[
    {
        "name": "CopyFromAzurePostgreSql",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<AzurePostgreSql input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "AzurePostgreSqlSource",
                "query": "<custom query e.g. SELECT * FROM mytable>",
                "queryTimeout": "00:10:00"
            },
            "sink": {
                "type": "<sink type>"
            }
        }
    }
]

将数据复制到 Azure Database for PostgreSQL 时，复制活动的 sink 节支持以下属性：

示例：

"activities":[
    {
        "name": "CopyToAzureDatabaseForPostgreSQL",
        "type": "Copy",
        "inputs": [
            {
                "referenceName": "<input dataset name>",
                "type": "DatasetReference"
            }
        ],
        "outputs": [
            {
                "referenceName": "<Azure PostgreSQL output dataset name>",
                "type": "DatasetReference"
            }
        ],
        "typeProperties": {
            "source": {
                "type": "<source type>"
            },
            "sink": {
                "type": "AzurePostgreSqlSink",
                "preCopyScript": "<custom SQL script>",
                "writeMethod": "CopyCommand",
                "writeBatchSize": 1000000
            }
        }
    }
]

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

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

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

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

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

示例：具有物理分区的大型表的完全负载

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

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

"source": {
    "type": "AzurePostgreSqlSource",
    "query": "SELECT * FROM <TableName> WHERE ?AdfDynamicRangePartitionCondition 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>"
    }
}

在映射数据流中转换数据时，可以从 Azure Database for PostgreSQL 读取表以及将数据写入表。 有关详细信息，请参阅映射数据流中的 源转换 和 汇聚转换 。 可以选择使用 Azure Database for PostgreSQL 数据集或 内联数据集 作为源和接收器类型。

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

使用 Azure Database for PostgreSQL 作为源类型时，关联的数据流脚本为：

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

下表列出了 Azure Database for PostgreSQL 接收器支持的属性。 您可以在“Sink选项”选项卡中编辑这些属性。

• 启用增量提取：使用此选项告知 ADF 仅处理自上次执行管道以来已更改的行。

• 增量日期列：使用增量提取功能时，必须选择要用作源表中水印的日期/时间或数字列。

• 从头开始读取：使用增量提取设置此选项将指示 ADF 在首次执行具有增量提取的管道时读取所有行。

使用 Azure Database for PostgreSQL 作为接收器类型时，关联的数据流脚本为：

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

有关属性的详细信息，请参阅 查找操作。

在“编辑链接服务”页中，在“版本”下选择 2.0，并通过引用链接服务属性版本 2.0 配置链接服务。

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