DataWorks:PostgreSQL data source

Offline read and write

• The PostgreSQL data source supports password-based authentication, including the SCRAM-SHA-256 method. If you change the password or authentication method in the PostgreSQL database, you must update the data source configuration, test the connection again, and manually run related tasks to verify the changes.

• PostgreSQL

• If a table or column name in PostgreSQL starts with a digit, is case-sensitive, or includes a hyphen (-), you must enclose the name in double quotation marks (""). Otherwise, the PostgreSQL plugin cannot read or write data. In the PostgreSQL Reader and Writer plugins, double quotation marks ("") are a JSON keyword. Therefore, you must escape the double quotation marks with a backslash (\). For example, if a table is named 123Test, the escaped name is \"123Test\".

  Note

  • Both the opening and closing double quotation marks must be escaped with a backslash (\).

  • Wizard mode does not support escaping. You must switch to script mode to escape characters.

  The following code shows an example of how to escape characters in script mode.

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

• Updating data in a PostgreSQL data source based on a unique index is not supported. To update data, you must first write the data to a temporary table and then use the RENAME operation.

Real-time read

The following limits apply to real-time sync tasks in Data Integration:

• Data Integration provides special support for ADD COLUMN:

  • Constraint: You cannot combine an ADD COLUMN operation with other ADD COLUMN, DROP COLUMN, or other DDL statements in a single transaction.

    Important

    If you use ADD COLUMN with other operations, such as DROP COLUMN、RENAME COLUMN or other ALTER COLUMN operations, the data synchronization task may fail to run properly.

  • Limitation: DDL operations other than ADD COLUMN cannot be recognized.

• ALTER TABLE/CREATE TABLE is not supported.

• Replication of TEMPORARY tables, UNLOGGED tables, and Hyper tables is not supported. PostgreSQL does not provide a mechanism to subscribe to WAL log parsing for these types of tables.

• Replication of Sequences (serial/bigserial/identity) is not supported.

• The TRUNCATE operation is not supported.

• Large object replication (Bytea) is not supported.

• Replication of views, materialized views, and foreign tables is not supported.

• When PostgreSQL is used as the source for single-table real-time or full-database real-time synchronization, only tables for which the account has owner permissions can be synchronized.

Preparation 1: Create an account and configure permissions

You must plan a database login account for subsequent operations. This account must have the REPLICATION、 LOGIN permissions on the database.

1. Create an account.

   For more information, see Create an account.

2. Configure permissions.

   Check whether the account has the replication permission.

   select userepl from pg_user where usename='xxx'

   The expected result is True. If the result is False, the account does not have the permission. You can run the following statement to grant the permission.

   ALTER USER <user> REPLICATION;

Preparation 2: Check whether a standby database is supported

SELECT pg_is_in_recovery()

Only the primary database is supported. The expected result is False. If the result is True, the database is a standby database. Real-time synchronization does not support standby databases. You must change the data source configuration to point to the primary database. For more information, see Configure the data source.

Preparation 3: Check whether wal_level is set to logical

show wal_level

wal_level specifies the level of wal_log. The expected result is logical. Otherwise, the logical replication mechanism is not supported.

Preparation 4: Check whether a wal_sender process can be started

-- Query max_wal_senders
show max_wal_senders;

-- Query the number of pg_stat_replication entries
select count(*) from pg_stat_replication

If max_wal_senders is not empty and its value is greater than the number of max_wal_senders entries in pg_stat_replication, idle wal_sender processes are available. The PostgreSQL database starts wal_sender processes for the data synchronization program to send logs to subscribers.

For each table that needs to be synchronized, you must manually run the ALTER TABLE [tableName] REPLICA IDENTITY FULL statement to grant the required permissions. Otherwise, the real-time synchronization task will fail.

After a PostgreSQL real-time synchronization task is started, slots and publications are automatically created in the database. The slot name format is di_slot_ + Solution ID , and the publication name format is di_pub_ + Solution ID. After the real-time synchronization task is stopped or undeployed, you must manually delete the slots and publications. Otherwise, PostgreSQL WAL files may continuously grow.

Single-table batch synchronization task configuration guide

• For more information about the procedure, see Configure a batch synchronization task in wizard mode and Configure a batch synchronization task in script mode.

• For all parameters and script demos for script mode configuration, see Appendix 1: Script demo and parameter description below.

Full-database batch read and full-database real-time read synchronization task configuration guide

For more information about the procedure, see Configure a full-database batch synchronization task and Configure a full-database real-time synchronization task.

Configure a batch synchronization task by using the code editor

If you want to configure a batch synchronization task by using the code editor, you must configure the related parameters in the script based on the unified script format requirements. For more information, see Script mode configuration. The following information describes the parameters that you must configure for data sources when you configure a batch synchronization task by using the code editor.

Reader script demo

To configure a job that extracts data from a PostgreSQL database, use script mode. For more information, see Configure a batch synchronization task in script mode.

{
    "type":"job",
    "version":"2.0",// The version number.
    "steps":[
        {
            "stepType":"postgresql",// The plugin name.
            "parameter":{
                "datasource":"",// The data source.
                "column":[// The columns.
                    "col1",
                    "col2"
                ],
                "where":"",// The filter condition.
                "splitPk":"",// The column used as the split key for data sharding. Data synchronization starts concurrent tasks for data synchronization.
                "table":""// The table name.
            },
            "name":"Reader",
            "category":"reader"
        },
        { 
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// The error count.
        },
        "speed":{
            "throttle":true, // A value of false for throttle indicates that throttling is disabled and the mbps parameter does not take effect. A value of true indicates that throttling is enabled.
            "concurrent":1, // The concurrency of the job.
            "mbps":"12"// The throttling rate. 1 mbps = 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Reader script parameters

Writer script demo

The following is a script configuration example. For more information, see the parameter descriptions above.

{
  "type":"job",
  "version":"2.0",// The version number.
  "steps":[ 
    {
      "stepType":"stream",
      "parameter":{},
      "name":"Reader",
      "category":"reader"
    },
    {
      "stepType":"postgresql",// The plugin name.
      "parameter":{
        "datasource":"",// The data source.
        "column":[// The columns.
          "col1",
          "col2"
        ],
        "table":"",// The table name.
        "preSql":[],// The SQL statements executed before the data synchronization task.
        "postSql":[],// The SQL statements executed after the data synchronization task.
      },
      "name":"Writer",
      "category":"writer"
    }
  ],
  "setting":{
    "errorLimit":{
      "record":"0"// The error count.
    },
    "speed":{
      "throttle":true,// A value of false for throttle indicates that throttling is disabled and the mbps parameter does not take effect. A value of true indicates that throttling is enabled.
      "concurrent":1, // The concurrency of the job.
      "mbps":"12"// The throttling rate. 1 mbps = 1 MB/s.
    }
  },
  "order":{
    "hops":[
      {
        "from":"Reader",
        "to":"Writer"
      }
    ]
  }
}

Writer script parameters

{
    "job":
    {
        "content":
        [{
            "reader": {...},
            "writer":
            {
                "parameter":
                {
                    "column":
                    [
                        // Destination table field list
                        "bigint_arr",
                        "double_arr",
                        "text_arr",
                        "jsonb_obj",
                        "json_obj"
                    ],
                    "pgType":
                    {
                        // Special type settings. The key is the field name of the destination table, and the value is the field type.
                        "bigint_arr": "bigint[]",
                        "double_arr": "double[]",
                        "text_arr": "text[]",
                        "jsonb_obj": "jsonb",
                        "json_obj": "json"
                    }

                }
            }
        }]
    }
}

PostgreSQL SSL authentication file description

When you create or modify a PostgreSQL data source connection in DataWorks, you can configure SSL authentication. The SSL authentication configuration parameters are described below.

Obtain PostgreSQL SSL authentication files

This section uses an ApsaraDB RDS for PostgreSQL instance as an example to generate SSL authentication certificates.

1. Obtain the Truststore certificate file.

   For more information about obtaining the Truststore certificate file, see Configure an SSL-encrypted connection.

   1. Go to the RDS instance list, click the RDS instance in the corresponding region, and then click the target instance ID to go to the instance details page.

   2. Select the connection string that you want to protect. The following figure shows the procedure:

   

   Note

   • If the public endpoint is enabled, both the internal and public endpoints are displayed. A cloud certificate can protect only one endpoint. Because the internal endpoint is relatively more secure, we recommend that you protect the public endpoint. For more information about how to view internal and public endpoints, see View internal and public endpoints.

   • If you need to protect both the internal and public endpoints, see Configure a custom certificate.

   • After you configure the cloud certificate, the Running Status of the instance changes to Modifying SSL. This status lasts about three minutes. Wait until the running status changes to Running before you proceed.

   c. Click Download CA Certificate to obtain the Truststore certificate file.

   

   The downloaded CA certificate package contains three files. When you configure the PostgreSQL data source in DataWorks, upload the file with the .pem extension or the file with the .p7b extension to the Truststore certificate file configuration item.

2. Obtain and configure the Keystore certificate file, Private key file, and Private key password.

   Prerequisites: You have completed configuring an SSL-encrypted connection or configuring a custom certificate, and you have the OpenSSL tool.

   Note

   If you use a Linux system, OpenSSL is pre-installed. If you use a Windows system, download and install the OpenSSL software package.

   For more information about obtaining and configuring the Keystore certificate file, Private key file, and Private key password, see Configure a client certificate.

   1. Use the OpenSSL tool on a Linux system or install the OpenSSL software on a Windows system to generate a self-signed certificate (ca1.crt) and its private key (ca1.key).

      openssl req -new -x509 -days 3650 -nodes -out ca1.crt -keyout ca1.key -subj "/CN=root-ca1"

   2. Generate a client certificate signing request file (client.csr) and a client certificate private key (client.key).

      openssl req -new -nodes -text -out client.csr -keyout client.key -subj "/CN=<client_username>"

      In this command, the CN value after the -subj parameter must be set to the username used by the client to access the database.

   3. Generate the client certificate (client.crt).

      openssl x509 -req -in client.csr -text -days 365  -CA ca1.crt -CAkey ca1.key -CAcreateserial  -out client.crt

   4. If your RDS PostgreSQL server requires client CA certificate verification, open the generated client self-signed certificate ca1.crt file, copy the certificate content, and paste it into the Client CA Certificate Public Key Content dialog as the client CA certificate.

   5. After configuring the client CA certificate on the RDS side, when you configure the PostgreSQL data source in DataWorks, convert the client certificate private key client.key to a client.pk8 file, and upload the client.pk8 file to the Private key file configuration item in the DataWorks PostgreSQL data source configuration.

      cp client.key client.pk8

   6. Configure the private key password.

      openssl pkcs8 -topk8 -inform PEM -in client.key -outform der -out client.pk8 -v1 PBE-MD5-DES

      Note

      When you run the command to configure the private key password, you must enter a password. If you set a password, use the same password for the private key password in the DataWorks PostgreSQL data source configuration.

Configure PostgreSQL SSL authentication files

When you upload the obtained certificate files to the DataWorks PostgreSQL configuration items, perform the following operations:

• Truststore certificate file: Upload the file with the .pem extension or the file with the .p7b extension obtained in the Obtain the Truststore certificate file step.

• Keystore certificate file: Upload the client certificate file client.crt obtained in the Generate the client certificate step.

• Private key file: Upload the client.pk8 file converted from the client certificate private key client.key in the Convert the private key file step.

• Private key password: The password configured in the Configure the private key password step.

ACL configuration: Go to the RDS instance list, click the RDS instance in the corresponding region, click the target instance ID to go to the instance details page, and then click Data Security > ACL Configuration to modify the setting. You can select different SSL authentication methods. For more information, see Configure ACL.