PostgreSQL 是一种常用的关系型数据库，数据集成同步任务为您提供读取和写入 PostgreSQL 数据源的双向通道能力。
本文为您介绍 DataSail 的 PostgreSQL 数据源配置、同步任务可视化和脚本模式（DSL）配置能力，实现与不同数据源的数据互通能力。

1 支持的 PostgreSQL 版本

• 离线读写
  目前支持读写的 PostgreSQL 可选版本为 PostgreSQL 10、11、12、13、14。
  您可以在数据库中执行以下语句，查看 PostgreSQL 数据库的版本：

  show server_version
  

2 使用前提

1. 子账号新建数据源时，需要有项目的管理员角色，方可以进行新建数据源操作。各角色对应权限说明，详见：管理成员

2. 确保集成同步任务使用的独享数据集成资源组，具有 PostgreSQL 数据库节点的网络访问能力。网络互通方案详见网络连通解决方案。

   • 数据源为 RDS 云数据库实例时，需要将集成资源组所在 VPC 中的 IPv4 CIDR 地址添加到 PostgreSQL 访问白名单中：
     1. 确认集成资源组所在的 VPC：
        
     2. 查看 VPC 的 IPv4 CIDR 地址：

        注意

        若考虑安全因素，减少 IP CIDR 的访问范围，您至少需要将集成资源组绑定的子网下的 IPv4 CIDR 地址加入到数据库白名单中。

        
     3. 将获取到的 IPv4 CIDR 地址添加进 PostgreSQL 数据库白名单中，添加操作详见创建白名单。
        
   • 数据源为公网自建数据源，需通过公网形式访问：
     1. 集成资源组开通公网访问能力，操作详见开通公网。
     2. 并将公网 IP 地址，添加进 PostgreSQL 数据库白名单中。

3. 数据源配置时填写的用户名信息，需确保有 SuperUser 或 replication 角色权限：

   • SuperUser：
     PostgreSQL 实时整库解决方案执行时，需要拥有 SuperUser 高级权限的用户账号来创建 publication，权限检查及设置 SQL 如下：

     --查看当前用户账号是否有 usesuper，返回False则表示无权限
     SELECT usesuper FROM pg_user WHERE usename = 'your_username' 
     --若当前账号无高权限，则需请拥有高权限的账号将您的账号添加为 usesuper
     ALTER USER your_username WITH SUPERUSER;
     

     如果数据源配置的用户名无法授予高权限，那么需要请有高权限的账号，来手动创建 publication，创建命令如下：

     # 默认是订阅全部表，这样支持新增表可以自动添加进来
     CREATE PUBLICATION dbz_publication FOR ALL TABLES
     

   • replication：
     replication 权限是为 PostgreSQL 数据库复制功能设计的权限。复制机制是用于实现数据实时同步、数据分发和高可用性的重要特性，拥有 replication 权限的用户，允许进行表的逻辑复制、管理发布和订阅等相关操作。

     --检查账号是否有replication权限,预期返回结果为True，返回False则表示无权限
     select userepl from pg_user where usename='xxx'
     
     --没有replication权限时，可以通过如下语句进行授权：
     ALTER USER <user> REPLICATION;
     

4. 为了支持 PostgreSQL 实时整库解决方案在全量同步时读取从库，增量同步时读取主库（Binlog 必须从主库读取），需要在 PostgreSQL 数据源配置时，新增高级参数 slave_host，其对应参数值为从库域名信息。

   说明

   该数据源建议只在解决方案场景中使用。

   

5. PostgreSQL 数据源配置实时整库时，需将数据库集群中 wal_level 设置为 logical，支持逻辑复制机制，参数设置完成后，需重启集群生效。以下为火山引擎 PostgreSQL 实例配置截图。
   

6. 检查是否可以启动wal_sender进程

   -- 查询 max_wal_senders
   show max_wal_senders;
   
   -- 查询 pg_stat_replication 数量
   select count(*) from pg_stat_replication
   

   当max_wal_senders返回值不为空，且其值大于pg_stat_replication值时，意味着存在有剩余且可用的wal_sender进程。此时，PostgreSQL 数据库将会为同步数据程序启动wal_sender进程来给订阅者发送日志。

7. PostgreSQL 数据库中复制标识必须设置为 Full 模式：

   --执行以下语句来检查是否已设置为 Full 模式：
   select relreplident from pg_class where relname like 'table_name(需替换为具体表名)';
   

   可能返回结果如下：

   • d：默认情况下，如果表有主键，复制标识就是主键；
   • n：表示不存在复制标识信息，此设置中，无法精准识别 update 和 delete 操作，对应数据可能不会实时同步到目标表中；
   • f：表示复制标识设置为所有列，update 和 delete 操作时，都能精准识别到每行每列。
     设置 Full 模式执行语句如下：

   ALTER TABLE table_name REPLICA IDENTITY FULL;
   

3 支持的字段类型

当前主要字段支持情况如下

4 数据同步任务开发

4.1 数据源注册

新建数据源操作详见配置数据源，下面为您介绍用连接串方式配置 PostgreSQL 数据源信息：

4.2 新建任务

PostgreSQL 数据源测试连通性成功后，进入到数据开发界面，开始新建 PostgreSQL 相关通道任务。
新建任务方式详见离线数据同步、流式数据同步。

4.3 可视化配置说明

任务创建成功后，您可根据实际场景，配置PostgreSQL 批式读、PostgreSQL 批式写或 PostgreSQL 流式写等通道任务。

4.3.1 PostgreSQL 批式读

数据来源选择 PostgreSQL，并完成以下相关参数配置：
其中参数名称前带 * 的为必填参数，名称前未带 * 的为可选填参数。

4.3.2 PostgreSQL 批式写

数据来源选择 PostgreSQL，并完成以下相关参数配置：
其中参数名称前带 * 的为必填参数，名称前未带 * 的为可选填参数。

4.3.3 PostgreSQL 流式写

支持可视化方式配置流式写入 PostgreSQL 单表。PostgreSQL Writer 通过 JDBC 远程连接 PostgreSQL 数据库，并执行相应的 SQL 语句，将数据写入 PostgreSQL。流式写入 PostgreSQL 配置方式如下：

数据目标端选择 PostgreSQL，并完成以下相关参数配置：
其中参数名称前带 * 的为必填参数，名称前未带 * 的为可选填参数。

4.3.4 字段映射

数据来源和目标端配置完成后，需要指定来源和目标端的字段映射关系，根据字段映射关系，数据集成任务将源端字段中的数据，写入到目标端对应字段中。PostgreSQL 同时也支持您在源端配置字段自定义数据类型，来满足您更复杂的数据结构或业务特定场景。
字段映射支持选择基础模式和转换模式配置映射：

• 转换模式：
  字段映射支持数据转换，您可根据实际业务需求进行配置，将源端采集的数据，事先通过数据转换后，以指定格式输入到目标端数据库中。
  转换模式详细操作说明详见4.1 转换模式
  在转换模式中，你可依次配置：来源节点、数据转换、目标节点信息：

  配置节点	说明
  来源节点	配置数据来源 Source 节点信息： 节点名称：自定义输入来源节点名称信息，只允许由数字、字母、下划线、-和.组成；且长度不能超过10。 数据字段：通过自动添加、手动添加等方式添加数据来源字段信息。 配置完成后，单击确认按钮，完成来源节点配置。
  数据转换	单击数据转换右侧添加按钮，选择 SQL 转换方式，配置转换信息和规则： 节点名称：自定义输入来源节点名称信息，只允许由数字、字母、下划线、-和.组成；且长度不能超过10。 SQL 脚本：输入 SQL 脚本转换规则，目前仅支持添加一个转换的 SQL 语句，且不能包括 “;”。 配置完成后，单击确认按钮，完成数据转换节点配置。SQL 脚本示例详见4.1.2 添加转换节点。
  目标节点	配置目标节点 Sink 信息： 节点名称：自定义输入来源节点名称信息，只允许由数字、字母、下划线、-和.组成；且长度不能超过10。 数据字段：通过自动添加、手动添加等方式添加数据目标字段信息。 配置完成后，单击确认按钮，完成目标节点配置。

• 基础模式：

  您可通过以下三种方式操作字段映射关系：

  • 自动添加：单击自动添加按钮，根据两端数据表信息，可以自动填充来源和目标的字段信息。
  • 手动添加：单击手动添加按钮，可以手动编辑来源和目标的字段信息，可以逐个添加。

    说明

    来源端字段信息支持输入数据库函数和常量配置，单击手动添加按钮，在源表字段中输入需添加的值，并选择函数或常量类型，例如：

    • 函数：支持您输入 now()、current_timestamp() 等 PostgreSQL 数据库支持的函数。
    • 常量：您可自定义输入常量值，'123'、'${DATE}'、'${hour}' 等，输入值两侧需要加上英文单引号，支持结合时间变量参数使用。
  • 移动\删除字段：您也可以根据需要移动字段映射顺序或删除字段。

4.4 DSL 配置说明

PostgreSQL 数据源支持使用脚本模式（DSL）的方式进行配置。
在某些复杂场景下，或当数据源类型暂不支持可视化配置时，您可通过任务脚本的方式，按照统一的 Json 格式，编写 PostgreSQL Reader 和 PostgreSQL Writer 参数脚本代码，来运行数据集成任务。

4.4.1 进入 DSL 模式

进入 DSL 模式操作流程，可详见 MySQL 数据源-4.4.1 进入DSL 模式。

4.4.2 PostgreSQL 批式读

进入 DSL 模式编辑界面后，您可根据实际情况替换相应参数，PostgreSQL 批式读脚本示例如下：

// 变量使用规则如下:
// 1.自定义参数变量: {{}}, 比如{{number}}
// 2.系统时间变量${}, 比如 ${date}、${hour}
// **************************************
{
    // [required] dsl version, suggest to use latest version
    "version": "0.2",
    // [required] execution mode, supoort streaming / batch now
    "type": "batch",
    // reader config
    "reader": {
        // [required] datasource type
        "type": "pg",
        // [optional] datasource id, set it if you have registered datasource
        "datasource_id": 12345,
        // [required] user parameter
        "parameter": {
            // ********** please write here **********
            // "key" : value
            "columns": [
                {
                    "name": "name_sample",
                    "type": "type_sample"
                }
            ],
            "filter": "id > 10",
            "split_pk": "split_pk_sample",
            "table_schema":"table_schema",
            "table_name": "table_name_sample"
        }
    },
    // writer config
    "writer": {
    },
    // common config
    "common": {
        // [required] user parameter
        "parameter": {
            // ********** please write here **********
            // "key" : value
        }
    }
}

Reader 参数说明，其中参数名称前带 * 的为必填参数，名称前未带 * 的为可选填参数：

4.4.3 PostgreSQL 批式写

根据实际情况替换 PostgreSQL 批式写相应参数，PostgreSQL 批式写脚本示例如下：

// **************************************
// 变量使用规则如下:
// 1.自定义参数变量: {{}}, 比如{{number}}
// 2.系统时间变量${}, 比如 ${date}、${hour}
// **************************************
{
    // [required] dsl version, suggest to use latest version
    "version": "0.2",
    // [required] execution mode, supoort streaming / batch now
    "type": "batch",
    // reader config
    "reader": {
        // [required] datasource type
        "type": "xx",
        // [optional] datasource id, set it if you have registered datasource
        "datasource_id": null,
        // [required] user parameter
        "parameter": {
            // ********** please write here **********
            // "key" : value
          
        }
    },
    // writer config
    "writer": {
        // [required] datasource type
        "type": "pg",
        // [optional] datasource id, set it if you have registered datasource
        "datasource_id": 12345,
        // [required] user parameter
        "parameter": {
            // ********** please write here **********
            // "key" : value
            "table_schema":"table_schema",
            "table_name":"table_1",
            "pre_sql_list":[""],
            "post_sql_list":[""],
            "write_mode":"directlyInsert",
            "columns": [
                {
                    "name": "name_sample",
                    "type": "type_sample"
                }
            ]
        }
    },
    // common config
    "common": {
        // [required] user parameter
        "parameter": {
            // ********** please write here **********
            // "key" : value
        }
    }
}

Writer 参数说明，其中参数名称前带 * 的为必填参数，名称前未带 * 的为可选填参数：

4.5 高级参数说明

• 对于可视化通道任务，读参数需要加上 job.reader. 前缀，写参数需要加上 job.writer. 前缀，如下图所示：
  
• 对于 DSL 任务，读参数请配置到 reader.parameter 下，写参数请配置到 writer.parameter 下，直接输入参数名称和参数值。如下图所示：
  

4.5.1 PostgreSQL 批式读

4.5.2 PostgreSQL 批式写