使用 Azure Data Factory 或 Synapse Analytics 在Azure Database for PostgreSQL中复制和转换数据

适用于： Azure Data Factory Azure Synapse Analytics

本文概述了如何在 Azure Data Factory 和 Synapse Analytics 管道中使用复制活动从/向Azure Database for PostgreSQL复制数据。 此外，如何使用Data Flow转换Azure Database for PostgreSQL中的数据。 若要了解详细信息，请阅读 Azure Data Factory 和 Synapse Analytics。

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

支持的功能

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

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

这三个活动适用于 Azure Database for PostgreSQL Single Server、Flexible Server 和 Azure Cosmos DB for PostgreSQL。

入门

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

• 复制数据工具
• Azure 门户
• .NET SDK
• Python SDK
• Azure PowerShell
• REST API
• Azure Resource Manager 模板

使用用户界面创建一个连接到 Azure Database for PostgreSQL 的链接服务

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

1. 浏览到Azure Data Factory或 Synapse 工作区中的“管理”选项卡，然后选择“链接服务”，然后选择“新建”

   • Azure Data Factory
   • Azure Synapse

   

2. 搜索 PostgreSQL 并选择 postgreSQL 连接器的 Azure 数据库。

   

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

   

连接器配置详细信息

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

链接服务属性

Azure Database for PostgreSQL 连接器版本2.0支持传输层安全性（TLS）1.3和多种安全套接字层（SSL）模式。 请参阅此 段落，将您的 Azure SQL Database 连接器版本从版本 1.0 升级。 有关属性详细信息，请参阅相应部分。

• 版本 2.0
• 版本 1.0

版本 2.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>" 
            }
        }
    }
}

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

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

若要使用系统分配的托管标识，请执行以下步骤：

1. 数据工厂或 Synapse 工作区可与系统分配的托管标识相关联。 了解详细信息， 生成系统分配的托管标识

2. 具有系统分配的托管标识的 PostgreSQL Azure 数据On。

3. 在 Security 下的 Azure Database for PostgreSQL 资源中

   1. 选择 身份验证

   2. 仅选择Microsoft Entra身份验证或PostgreSQL 和Microsoft Entra身份验证身份验证方法。

   3. 选择 + 添加 Microsoft Entra 管理员

   4. 将Azure Data Factory资源的系统分配托管标识添加为 Microsoft Entra 管理员之一

      

4. 为 PostgreSQL 链接服务配置Azure数据库。

示例：

{
    "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,
            "authenticationType": "SystemAssignedManagedIdentity"
        }
    }
}

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

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

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

你也需要执行以下步骤：

1. 请确保在 Azure 门户中创建 用户分配的托管标识资源。 若要了解详细信息，请转到“管理用户分配的托管标识”

2. 将用户分配托管标识分配给适用于 PostgreSQL 的 Azure 数据库资源

   1. 在 Azure 中的 PostgreSQL 服务器资源中，Security 下

   2. 选择 身份验证

   3. 验证身份验证方法是否是仅 Microsoft Entra 身份验证 还是 PostgreSQL 和 Microsoft Entra 身份验证

   4. 选择+ 添加 Microsoft Entra 管理员链接，然后选择用户分配的托管标识

      

3. 将 用户分配的托管标识分配给 Azure Data Factory 资源

   1. 选择“设置”，然后选择“托管标识”

   2. 在 “用户分配 ”选项卡下。选择 “+ 添加 ”链接并选择用户托管标识

      Azure 数据工厂资源中用户分配的托管标识配置的屏幕截图。

4. 为 PostgreSQL 链接服务配置Azure数据库。

示例：

{
    "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,
            "authenticationType": "UserAssignedManagedIdentity",
            "credential": {
                "referenceName": "<your credential>",
                "type": "CredentialReference"
            }
        }
    }
}

服务主体身份验证

示例：

服务主体密钥

{
    "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

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

典型的connection string是 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 作为数据源

若要从 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 作为目标端

若要将数据复制到 Azure Database for PostgreSQL，请将复制活动中的接收器类型设置为 SqlSink。 复制活动 接收槽 部分中支持的属性如下：

示例 1：复制命令

"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
            }
        }
    }
]

示例 2：更新或插入数据

"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",
                "writeMethod": "Upsert",
                "upsertSettings": {
                    "keys": [
                        "<column name>"
                    ]
                },
            }
        }
    }
]

更新插入数据

Copy activity 原生支持 upsert 操作。 为了执行 upsert 操作，用户应提供作为主键或唯一列的键列。 如果用户不提供键列，则使用接收器表中的主键列。 复制活动将在接收器表中更新非键列，其中键列值与源表中的列匹配;否则，它将插入新数据。

从Azure Database for PostgreSQL并行复制

复制活动中的Azure Database for PostgreSQL连接器提供内置的数据分区来并行复制数据。 可以在复制活动的 源 选项卡上找到数据分区选项。

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

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

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

1. 选择独特的列作为分区列（如主键或唯一键），以避免数据倾斜。
2. 如果表具有内置分区，请使用名为“表的物理分区”分区选项来提升性能。
3. 如果使用Azure Integration Runtime复制数据，则可以设置更大的“Data Integration Units （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源脚本示例

将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汇聚脚本示例

将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

脚本活动

有关脚本活动的详细信息，请参阅 脚本活动。

查找活动属性

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

升级 Azure Database for PostgreSQL 连接器

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

相关内容

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