DataWorks:PostgreSQL data source

Supported PostgreSQL versions

The following versions are supported: PostgreSQL 10, PostgreSQL 11, PostgreSQL 12, PostgreSQL 13, PostgreSQL 14, and PostgreSQL 15. You can execute the following statement to view the version of a PostgreSQL database:

show server_version

Limits

Batch data read and write

PostgreSQL data sources support password authentication and SCRAM-SHA-256 authentication. If the password and authentication method of the PostgreSQL database that you add to DataWorks as a data source are changed, you must update the configurations of the data source and test the network connectivity between the data source and your resource group again. In addition, you must also manually run the related data synchronization task again to ensure that the configurations of the task meet your business requirements.

"parameter": {
    "datasource": "abc",
    "column": [
        "id",
        "\"123Test\"", // Add escape characters.
],
"where": "",
"splitPk": "id",
"table": "public.wpw_test"
},

Real-time data read

Data type mappings

Batch data read and write

Most PostgreSQL data types are supported. Make sure that the data types of your database are supported.

Prepare a PostgreSQL environment before data synchronization

Before you use DataWorks to synchronize data from or to a PostgreSQL data source, you must prepare a PostgreSQL environment. This ensures that a data synchronization task can be configured and can synchronize data from or to the PostgreSQL data source as expected. The following information describes how to prepare a PostgreSQL environment for data synchronization from or to a PostgreSQL data source.

Preparation 1: Create an account and grant the required permissions to the account

You must create an account that is used to log on to the PostgreSQL database for subsequent operations. The account must have the REPLICATION and LOGIN permissions on the PostgreSQL database.

1. Create an account.

   For more information, see Create a database and an account on an ApsaraDB RDS for PostgreSQL instance.

2. Grant the required permissions to the account.

   Execute the following statement to check whether the account has the REPLICATION permission on the PostgreSQL database:

   select userepl from pg_user where usename='xxx'

   The expected return value is True. If False is returned, the account does not have the REPLICATION permission on the PostgreSQL database. You can execute the following statement to grant the REPLICATION permission to the account:

   ALTER USER <user> REPLICATION;

Preparation 2: Check whether the PostgreSQL database is a secondary database

SELECT pg_is_in_recovery()

Real-time synchronization supports only primary databases. The expected return value is False. If True is returned, the PostgreSQL database is a secondary database. Real-time synchronization does not support secondary databases. In this case, you must change the database that is configured in the PostgreSQL data source to the primary database. For more information, see Add a PostgreSQL data source.

Preparation 3: Check whether the value of the wal_level parameter is logical

show wal_level

The wal_level parameter specifies the wal_log level. The expected return value is logical. If logical is not returned, the logical replication mechanism is not supported.

Preparation 4: Check whether the wal_sender process can be started

-- Query the value of the max_wal_senders parameter.
show max_wal_senders;

-- Query the number of wal_sender processes in the pg_stat_replication view.
select count(*) from pg_stat_replication

If the max_wal_senders parameter is not empty and the value of the max_wal_senders parameter is greater than the number of wal_sender processes in the pg_stat_replication view, idle wal_sender processes exist. The PostgreSQL database starts a wal_sender process for the data synchronization program to send logs to subscribers.

Execute the ALTER TABLE [tableName] REPLICA IDENTITY FULL statement to enable logical replication for the table that you want to synchronize. If you do not execute the statement, an error may be reported for the real-time synchronization task.

If the task that is used to synchronize data from PostgreSQL in real time is stopped or undeployed, delete the task. After the task is started, the system creates slots and publications in the PostgreSQL database. The names of the slots are in the following format: di_slot_ + Task ID, and the names of the publications are in the following format: di_pub_ + Task ID. If you do not delete the task after the task is stopped or undeployed, the number of write-ahead logging (WAL) files in the PostgreSQL database will constantly increase.

Develop a data synchronization task

For information about the entry point for and the procedure of configuring a data synchronization task, see the following sections. For information about the parameter settings, view the infotip of each parameter on the configuration tab of the task.

Add a data source

Before you configure a data synchronization task to synchronize data from or to a specific data source, you must add the data source to DataWorks. For more information, see Add and manage data sources.

Configure a batch synchronization task to synchronize data of a single table

• For more information about the configuration procedure, see Configure a batch synchronization task by using the codeless UI and Configure a batch synchronization task by using the code editor.

• For information about all parameters that are configured and the code that is run when you use the code editor to configure a batch synchronization task, see Appendix: Code and parameters.

Configure synchronization settings to implement batch synchronization of all data in a database or real-time synchronization of full and incremental data in a single table or a database

For more information about the configuration procedure, see Configure a synchronization task in Data Integration.

Additional information

• Data synchronization between primary and secondary databases

  A secondary PostgreSQL database can be deployed for disaster recovery. The secondary database continuously synchronizes data from the primary database. Data latency between the primary and secondary databases cannot be prevented. This may result in data inconsistency.

• Data consistency control

  PostgreSQL is a relational database management system (RDBMS) that supports strong consistency for data queries. A database snapshot is created before a synchronization task starts. PostgreSQL Reader reads data from the database snapshot. Therefore, if new data is written to the database during data synchronization, PostgreSQL Reader cannot obtain the new data.

  Data consistency cannot be ensured if you enable PostgreSQL Reader to use parallel threads to read data in a synchronization task.

  PostgreSQL Reader shards the source table based on the value of the splitPk parameter and uses parallel threads to read data. These parallel threads belong to different transactions and read data at different points in time. Therefore, the parallel threads observe different snapshots.

  Data inconsistencies cannot be prevented if parallel threads are used for a synchronization task. The following workarounds can be used:

  • Enable PostgreSQL Reader to use a single thread to read data in a synchronization task. This indicates that you do not need to specify a shard key for PostgreSQL Reader. This way, data consistency is ensured, but data is synchronized at low efficiency.

  • Make sure that no data is written to the source table during data synchronization. This ensures that the data in the source table remains unchanged during data synchronization. For example, you can lock the source table or disable data synchronization between primary and secondary databases. This way, data is efficiently synchronized, but your ongoing services may be interrupted.

• Character encoding

  A PostgreSQL database supports only the EUC_CN and UTF-8 encoding formats for simplified Chinese characters. PostgreSQL Reader uses JDBC to read data. This enables PostgreSQL Reader to automatically convert the encoding formats of characters. Therefore, you do not need to specify the encoding format.

  If you specify the encoding format for a PostgreSQL database but data is written to the PostgreSQL database in a different encoding format, PostgreSQL Reader cannot identify this inconsistency and may export garbled characters.

• Incremental data synchronization

  PostgreSQL Reader uses JDBC to connect to a database and uses a SELECT statement with a WHERE clause to read incremental data.

  • For batch data, incremental add, update, and delete operations (including logically delete operations) are distinguished by timestamps. Specify the WHERE clause based on a specific timestamp. The time indicated by the timestamp must be later than the time indicated by the latest timestamp in the previous synchronization.

  • For streaming data, specify the WHERE clause based on the ID of a specific record. The ID must be greater than the maximum ID involved in the previous synchronization.

  If the data that is added or modified cannot be distinguished, PostgreSQL Reader can read only full data.

• Syntax validation

  PostgreSQL Reader allows you to configure custom SELECT statements by using the querySql parameter but does not verify the syntax of these statements.

Appendix: Code and parameters

Appendix: Configure a batch synchronization task by using the code editor

If you use the code editor to configure a batch synchronization task, you must configure parameters for the reader and writer of the related data source based on the format requirements in the code editor. For more information about the format requirements, see Configure a batch synchronization task by using the code editor. The following information describes the configuration details of parameters for the reader and writer in the code editor.

Code for PostgreSQL Reader

{
    "type":"job",
    "version":"2.0",// The version number. 
    "steps":[
        {
            "stepType":"postgresql",// The plug-in name. 
            "parameter":{
                "datasource":"",// The name of the data source. 
                "column":[// The names of the columns. 
                    "col1",
                    "col2"
                ],
                "where":"",// The WHERE clause. 
                "splitPk":"",// The shard key based on which the table is sharded. Data Integration runs parallel threads to synchronize data. 
                "table":""// The name of the table. 
            },
            "name":"Reader",
            "category":"reader"
        },
        { 
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// The maximum number of dirty data records allowed. 
        },
        "speed":{
            "throttle":true, // Specifies whether to enable throttling. The value false indicates that throttling is disabled, and the value true indicates that throttling is enabled. The mbps parameter takes effect only when the throttle parameter is set to true. 
            "concurrent":1, // The maximum number of parallel threads. 
            "mbps":"12"// The maximum transmission rate. Unit: MB/s. 
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Parameters in code for PostgreSQL Reader

Code for PostgreSQL Writer

{
  "type":"job",
  "version":"2.0",// The version number. 
  "steps":[ 
    {
      "stepType":"stream",
      "parameter":{},
      "name":"Reader",
      "category":"reader"
    },
    {
      "stepType":"postgresql",// The plug-in name. 
      "parameter":{
        "datasource":"",// The name of the data source. 
        "column":[// The names of the columns. 
          "col1",
          "col2"
        ],
        "table":"",// The name of the table. 
        "preSql":[]// The SQL statement that you want to execute before the synchronization task is run. 
        "postSql":[],// The SQL statement that you want to execute after the synchronization task is run. 
      },
      "name":"Writer",
      "category":"writer"
    }
  ],
  "setting":{
    "errorLimit":{
      "record":"0"// The maximum number of dirty data records allowed.
    },
    "speed":{
      "throttle":true,// Specifies whether to enable throttling. The value false indicates that throttling is disabled, and the value true indicates that throttling is enabled. The mbps parameter takes effect only when the throttle parameter is set to true. 
      "concurrent":1, // The maximum number of parallel threads. 
      "mbps":"12"// The maximum transmission rate. Unit: MB/s. 
    }
  },
  "order":{
    "hops":[
      {
        "from":"Reader",
        "to":"Writer"
      }
    ]
  }
}

Parameters in code for PostgreSQL Writer

{
    "job":
    {
        "content":
        [{
            "reader": {...},
            "writer":
            {
                "parameter":
                {
                    "column":
                    [
                        // The names of the columns to which you want to write data.
                        "bigint_arr",
                        "double_arr",
                        "text_arr",
                        "jsonb_obj",
                        "json_obj"
                    ],
                    "pgType":
                    {
                        // The configuration that is used to convert data types. In each key-value pair, the key specifies the name of a field in the destination table, and the value specifies the data type of the field. 
                        "bigint_arr": "bigint[]",
                        "double_arr": "double[]",
                        "text_arr": "text[]",
                        "jsonb_obj": "jsonb",
                        "json_obj": "json"
                    }

                }
            }
        }]
    }
}