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.

• If a table or field 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 (\).

  • The codeless UI does not support escaping. You must switch to the code editor to escape characters.

  The following code shows an example of how to escape characters in the code editor.

  "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 Data Definition Language (DDL) statements in a single transaction.

    Important

    If you use ADD COLUMN with other operations, such as DROP COLUMN or RENAME COLUMN, which are ALTER COLUMN behaviors, the data synchronization task will fail.

  • Limit: Data Integration cannot detect DDL operations other than ADD COLUMN.

• The ALTER TABLE and CREATE TABLE operations are not supported.

• Replication of TEMPORARY, UNLOGGED, and Hyper tables is not supported. PostgreSQL does not provide a mechanism to parse logs and subscribe to these types of tables.

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

• The TRUNCATE operation is not supported.

• Replication of large objects (Bytea) is not supported.

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

Preparation 1: Create an account and configure permissions

Create a database logon account for subsequent operations. This account must have the REPLICATION and LOGIN permissions.

1. Create an account.

   For more information, see Create an account and a database.

2. Configure permissions.

   Check if the account has the replication permission.

   select userepl from pg_user where usename='xxx'

   If the command returns `True`, the permission is granted. If it returns `False`, the permission is not granted. To grant the permission, run the following statement:

   ALTER USER <user> REPLICATION;

Preparation 2: Check for secondary database support

SELECT pg_is_in_recovery()

Only primary databases are supported. The command must return `False`. If it returns `True`, the database is a secondary database. Real-time synchronization does not support secondary databases. You must change the data source configuration to use the primary database. For more information, see Configure a PostgreSQL data source.

Preparation 3: Check if wal_level is set to logical

show wal_level

The wal_level parameter specifies the level of the wal_log. The command must return `logical`. If it does not, the logical replication mechanism is not supported.

Preparation 4: Check if a wal_sender process can be started

-- Query max_wal_senders.
show max_wal_senders;

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

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

For each table that you want to synchronize, you must run the ALTER TABLE [tableName] REPLICA IDENTITY FULL statement to grant the required permissions. Otherwise, the real-time sync task reports an error.

After a real-time PostgreSQL sync task starts, it automatically creates a slot and a publication in the database. The slot name is in the format di_slot_ + Solution ID, and the publication name is in the format di_pub_ + Solution ID. After the real-time sync task is stopped or unpublished, you must manually delete the slot and publication. Otherwise, the PostgreSQL Write-Ahead Logging (WAL) may continue to grow.

Guide to configuring offline sync tasks for a single table

• For more information, see Configure a task in the codeless UI and Configure a task in the code editor.

• For a complete list of parameters and a script example for the code editor, see Appendix 1: Script examples and parameter descriptions.

Guide to configuring full database offline read and real-time sync tasks

For more information, see Configure a real-time sync task for a full database.

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 Configuration in the code editor. 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 example

To configure a job that synchronizes and extracts data from a PostgreSQL database, see the instructions in Configure a task in the code editor.

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

Reader script parameters

Writer script example

The following code provides a script configuration example. For more information, see the parameter descriptions that follow.

{
  "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 fields.
          "col1",
          "col2"
        ],
        "table":"",// The table name.
        "preSql":[],// The SQL statement to execute before the data synchronization task runs.
        "postSql":[],// The SQL statement to execute after the data synchronization task runs.
      },
      "name":"Writer",
      "category":"writer"
    }
  ],
  "setting":{
    "errorLimit":{
      "record":"0"// The number of error records.
    },
    "speed":{
      "throttle":true,// If throttle is set to false, the mbps parameter does not take effect and throttling is disabled. If throttle is set to true, throttling is enabled.
      "concurrent":1, // The number of concurrent jobs.
      "mbps":"12"// The throttling rate. 1 mbps is equal to 1 MB/s.
    }
  },
  "order":{
    "hops":[
      {
        "from":"Reader",
        "to":"Writer"
      }
    ]
  }
}

Writer script parameters

{
    "job":
    {
        "content":
        [{
            "reader": {...},
            "writer":
            {
                "parameter":
                {
                    "column":
                    [
                        // List of fields in the destination table.
                        "bigint_arr",
                        "double_arr",
                        "text_arr",
                        "jsonb_obj",
                        "json_obj"
                    ],
                    "pgType":
                    {
                        // Special type settings. The key is the field name in 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 files

When you create or modify a connection for a PostgreSQL data source in DataWorks, you can configure SSL authentication. The following table describes the configuration items related to SSL authentication.

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 how to obtain the Truststore Certificate File, see Use a cloud certificate to quickly enable SSL encryption.

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

   2. Select the connection string to protect, as shown in the following figure:

   

   Note

   • If a public endpoint is enabled, the system displays both an internal endpoint and a public endpoint. A cloud certificate can protect only one endpoint. We recommend that you protect the public endpoint because it is exposed to the internet, whereas the internal endpoint is more secure within the VPC. For more information about how to view the internal and public endpoints, see View internal and public endpoints.

   • To protect both the internal and public endpoints, see Use a custom certificate to enable SSL encryption.

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

   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 or .p7b extension to the Truststore Certificate File field.

2. Obtain and configure the Keystore Certificate File, Private Key File, and Private Key Password.

   Prerequisites: You have completed the steps in Use a cloud certificate to quickly enable SSL encryption or Use a custom certificate to enable SSL encryption, and you have the OpenSSL tool installed.

   Note

   Linux systems include the OpenSSL tool by default. If you use a Windows system, you must obtain and install the OpenSSL package.

   For more information about how to obtain and configure the Keystore Certificate File, Private Key File, and Private Key Password, see Configure a client CA certificate.

   1. Use the OpenSSL tool to generate a self-signed certificate (`ca1.crt`) and a 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 (CSR) file (`client.csr`) and a client private key (`client.key`).

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

      In this command, set the Common Name (CN) value after the -subj parameter to the username that the client uses to access the database.

   3. Generate a 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 ApsaraDB RDS for PostgreSQL server needs to verify the client CA certificate, open the generated self-signed certificate file `ca1.crt`. Copy the certificate content and paste it into the Enter The Public Key Of The Client Certificate Authority dialog box.

   5. After you configure the client CA certificate on the RDS instance, you must convert the client private key file `client.key` to the PKCS#8 format (`client.pk8`). When you configure the PostgreSQL data source in DataWorks, upload the `client.pk8` file to the Private Key File field.

      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 are prompted to enter a password. If you set a password, you must use the same password for the Private Key Password field in the DataWorks PostgreSQL data source configuration.

Configure PostgreSQL SSL authentication files

When you upload the certificate files to the PostgreSQL data source configuration in DataWorks, follow these steps:

• Truststore Certificate File: Upload the .pem or .p7b file that you obtained in the Obtain the Truststore certificate file step.

• Keystore Certificate File: Upload the `client.crt` file that you obtained in the Generate a client certificate step.

• Private Key File: Upload the `client.pk8` file that you obtained in the Convert the private key file step.

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

Configure ACL: Go to the RDS Instances page, find the RDS instance in the target region, and then click the instance ID to go to the instance details page. Click Data Security > Configure ACL to modify the settings. You can choose from different SSL authentication methods. For more information, see Force clients to use SSL connections.