DataWorks:PostgreSQL

Offline read and write

Reading from views is supported.

• Authentication: Password-based authentication is supported, including SCRAM-SHA-256. If the password or authentication method changes in the PostgreSQL database, update the data source configuration, test the connection, and manually re-run related tasks.

• Special characters in table and field names: Table or field names that start with a digit, are case-sensitive, or contain a hyphen (-) must be enclosed in double quotation marks (""). Because double quotation marks are a JSON keyword in the PostgreSQL Reader and Writer plugins, escape each one with a backslash (\). For example, 123Test becomes \"123Test\". Example (code editor):

  - Escape both the opening and closing double quotation marks. - The codeless UI does not support escaping. Switch to the code editor to escape characters.

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

• Updates based on unique index: Not supported. Write data to a temporary table first, then use the RENAME operation.

Real-time read

The following limitations apply to real-time sync tasks in Data Integration.

Step 1: Create a database account and configure permissions

Create a database login account with REPLICATION and LOGIN permissions.

Real-time sync uses logical replication, which follows a publish-and-subscribe model. The subscriber pulls data changes from publications on the publisher node. Logical replication starts by copying a snapshot from the publisher database, then streams changes in real time.

1. Create a database account.

2. Verify that the account has the replication permission:

   SELECT userepl FROM pg_user WHERE usename='<username>';

   If the result is False, grant the permission:

   ALTER USER <username> REPLICATION;

Step 2: Confirm the database is a primary (real-time sync only)

Real-time sync supports only primary databases. Run the following to check:

SELECT pg_is_in_recovery();

The result must be False. If it returns True, the database is a standby. Update the data source configuration to use the primary database. For more information, see Configure a PostgreSQL data source.

Step 3: Verify that wal_level is set to logical (real-time sync only)

SHOW wal_level;

The result must be logical. If it is not, logical replication is not supported on this database.

Step 4: Verify that wal_sender processes are available (real-time sync only)

-- Check the maximum number of wal_sender processes allowed.
SHOW max_wal_senders;

-- Check how many are currently in use.
SELECT COUNT(*) FROM pg_stat_replication;

If max_wal_senders is greater than the count from pg_stat_replication, idle wal_sender processes are available. PostgreSQL starts a wal_sender process for each data synchronization program that sends logs to subscribers.

For each table you want to synchronize, run:

ALTER TABLE <table_name> REPLICA IDENTITY FULL;

This is required. Without it, the real-time sync task reports an error.

Offline sync (single table)

• Use the codeless UI

• Use the code editor

For a complete parameter reference and script example, see Appendix 1: Script examples and parameter descriptions.

Real-time sync and full database offline sync

See Configure a real-time sync task for a full database.

Reader script example

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "postgresql",
            "parameter": {
                "datasource": "",        // Data source name
                "column": [              // Columns to sync
                    "col1",
                    "col2"
                ],
                "where": "",             // Filter condition
                "splitPk": "",           // Column for data sharding (enables concurrent sync)
                "table": ""              // Table name
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"                // Maximum number of error records
        },
        "speed": {
            "throttle": true,            // true: throttling enabled; false: mbps has no effect
            "concurrent": 1,             // Number of concurrent jobs
            "mbps": "12"                 // Throttle rate (1 mbps = 1 MB/s)
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Reader parameters

Writer script example

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "postgresql",
            "parameter": {
                "datasource": "",        // Data source name
                "column": [              // Destination columns
                    "col1",
                    "col2"
                ],
                "table": "",             // Destination table name
                "preSql": [],            // SQL to run before the sync task
                "postSql": []            // SQL to run after the sync task
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "throttle": true,
            "concurrent": 1,
            "mbps": "12"
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Writer parameters

`pgType` example:

{
    "job": {
        "content": [{
            "reader": {},
            "writer": {
                "parameter": {
                    "column": [
                        "bigint_arr",
                        "double_arr",
                        "text_arr",
                        "jsonb_obj",
                        "json_obj"
                    ],
                    "pgType": {
                        "bigint_arr": "bigint[]",
                        "double_arr": "double[]",
                        "text_arr":   "text[]",
                        "jsonb_obj":  "jsonb",
                        "json_obj":   "json"
                    }
                }
            }
        }]
    }
}

SSL authentication files

When creating or modifying a PostgreSQL data source connection in DataWorks, configure SSL authentication based on your Access Control List (ACL) mode.

prefer: The PostgreSQL server does not forcibly verify the client certificate. If no SSL files are configured, a regular connection is used.

verify-ca: All three files are required. The server verifies the client certificate before establishing the connection.

Obtain the SSL authentication files

This section uses an ApsaraDB RDS for PostgreSQL instance as an example.

Configure SSL authentication files in DataWorks

Upload the certificate files when creating or editing the PostgreSQL data source:

• Truststore File: Upload the .pem or .p7b file from step 3 of Obtain the Truststore File.

• Keystore File: Upload client.crt from step 3 of Obtain the Keystore File, Private Key File, and Private Key Password.

• Private Key File: Upload client.pk8 from step 5 of Obtain the Keystore File, Private Key File, and Private Key Password.

• Private Key Password: The password set in step 6 (if applicable).

To set the Configure ACL mode: Go to the RDS Instances page, click the instance ID, then go to Data Security > Configure ACL. For more information, see Force clients to use SSL connections.

prefer: The PostgreSQL server does not forcibly verify the client certificate.

verify-ca: Upload the correct client certificate in DataWorks so the server can verify the client's identity.