DataWorks:Hologres data source

Batch read and write

• Hologres Writer does not support writing data to Hologres external tables.

• Logic for obtaining the Hologres endpoint for data source connectivity:

  • For Hologres instances in the current region, the endpoint priority is: any Tunnel > Single Tunnel > Public (internet).

  • For Hologres instances in other regions, the endpoint priority is: Public (internet) > Single Tunnel.

Single-table real-time read

• The Hologres version must be 2.1 or later.

• Incremental synchronization of Hologres partitioned tables is not supported.

• Synchronization of DDL change messages on Hologres tables is not supported.

• Hologres incremental synchronization supports the following data types:

  INTEGER, BIGINT, TEXT, CHAR(n), VARCHAR(n), REAL, JSON, SERIAL, OID, INT4[], INT8[], FLOAT8[], BOOLEAN[], TEXT[].

• For single-table real-time synchronization from Hologres, you must enable Hologres Binlog on the source Hologres table. For more information, see Subscribe to Hologres Binlog.

Full-database real-time write

• Real-time data synchronization tasks do not support synchronizing tables without primary keys.

• When synchronizing data from a MySQL full database to Hologres in real time, data can only be written to child tables of partitioned tables. Writing to the parent table of a partitioned table is not supported.

Batch read

Hologres Reader supports two read modes:

JDBC mode (default)

Reads data from Hologres tables through PSQL. Multiple concurrent tasks are initiated based on the shard count of the table, with each shard corresponding to one concurrent Select task:

• When creating a table in Hologres, you can configure the shard count of the table by running CALL set_table_property('table_name', 'shard_count', 'xx') within the same CREATE TABLE transaction.

  By default, the database uses the default shard count, which depends on the configuration of the Hologres instance.

• The Select statement filters data by shard using the built-in column hg_shard_id of the table.

Arrow mode (useArrow=true)

Reads data through the Hologres COPY OUT Arrow protocol. This mode supports LZ4 compressed transfer and provides higher performance:

• Requires Hologres version >= 4.0.18.

• Concurrent tasks are also split based on the shard count.

• Function columns and constant columns are not supported. Only physical columns in the table are supported.

• Compressed transfer is automatically enabled when the version requirement is met.

Batch write

Hologres Writer obtains protocol data generated by the Reader through the data synchronization framework. The channel and conflict resolution strategy used to write data are determined by the writeMode and conflictMode configurations.

Write mode (writeMode)

Scenario recommendations

Conflict handling mode (conflictMode)

You can configure conflictMode to specify how newly imported data is handled when a primary key conflict occurs with existing data:

• When conflictMode is set to Replace (full row update), new data overwrites old data. All columns in the row are overwritten. Columns that are not included in the column mapping are forcibly set to NULL.

• When conflictMode is set to Update, new data overwrites old data. Only the columns that are included in the column mapping are overwritten.

• When conflictMode is set to Ignore, the new data is ignored.

Single-table batch synchronization

• Supported data sources: All data source types supported by the Data Integration module.

• Configuration guide: Configure a batch synchronization task in the codeless UI and Configure a batch synchronization task by using the code editor. For the complete parameters and script demos available in the code editor, see Appendix: Complete parameters and script demos below.

Single-table real-time synchronization

• Supported data sources: DataHub, Hologres, Kafka, and LogHub

• Configuration guide: Configure a real-time synchronization task.

Full-database batch synchronization

• Supported data sources: AnalyticDB for MySQL 3.0, ClickHouse, Doris, Hologres, Oracle, PolarDB, and SQL Server

• Configuration guide: Configure a full-database batch synchronization task

Full-database real-time synchronization

• Supported data sources: AnalyticDB for OceanBase, MongoDB, MySQL, Oracle, PolarDB, PolarDB-X 2.0, and PostgreSQL

• Configuration guide: Configure a real-time synchronization task to synchronize data of a single table

Serverless full-database real-time

• Supported data source: MySQL

• Configuration guide: Configure a Serverless 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 demos

• Configure a non-partitioned table

  • The following example shows how to configure reading data from a Hologres non-partitioned table into memory.

    {
        "transform": false,
        "type": "job",
        "version": "2.0",
        "steps": [
            {
                "stepType": "holo",
                "parameter": {
                    "datasource": "holo_db",
                    "envType": 1,
                    "column": [ 
                        "tag",
                        "id",
                        "title",
                        "body"
                    ],
                    "where": "",
                    "table": "holo_reader_basic_src"
                },
                "name": "Reader",
                "category": "reader"
            },
            {
                "stepType": "stream",
                "parameter": {
                    "print": false,
                    "fieldDelimiter": ","
                },
                "name": "Writer",
                "category": "writer"
            }
        ],
        "setting": {
            "executeMode": null,
            "failoverEnable": null,
            "errorLimit": {
                "record": "0"
            },
            "speed": {
                "concurrent": 2,
                "throttle": false
            }
        },
        "order": {
            "hops": [
                {
                    "from": "Reader",
                    "to": "Writer"
                }
            ]
        }
    }

  • The DDL statement of the Hologres table is as follows.

    begin;
    drop table if exists holo_reader_basic_src;
    create table holo_reader_basic_src(
      tag text not null, 
      id int not null, 
      title text not null, 
      body text, 
      primary key (tag, id));
      call set_table_property('holo_reader_basic_src', 'orientation', 'column');
      call set_table_property('holo_reader_basic_src', 'shard_count', '3');
    commit;

• Configure a partitioned table

  • Configure reading data from a child table of a Hologres partitioned table into memory.

    Note

    Pay attention to the configuration of partition.

    {
        "transform": false,
        "type": "job",
        "version": "2.0",
        "steps": [
            {
                "stepType": "holo",
                "parameter": {
                    "selectedDatabase": "public",
                    "partition": "tag=foo",
                    "datasource": "holo_db",
                    "envType": 1,
                    "column": [
                        "tag",
                        "id",
                        "title",
                        "body"
                    ],
                    "tableComment": "",
                    "where": "",
                    "table": "public.holo_reader_basic_part_src"
                },
                "name": "Reader",
                "category": "reader"
            },
            {
          "stepType":"stream",
          "parameter":{},
          "name":"Writer",
          "category":"writer"
        }
      ],
      "setting":{
        "errorLimit":{
          "record":"0"
        },
        "speed":{
          "throttle":true,
          "concurrent":1,
          "mbps":"12"
            }
         },
      "order":{
        "hops":[
          {
            "from":"Reader",
            "to":"Writer"
          }
        ]
      }
    }

  • The DDL statement of the Hologres table is as follows.

    begin;
    drop table if exists holo_reader_basic_part_src;
    create table holo_reader_basic_part_src(
      tag text not null, 
      id int not null, 
      title text not null, 
      body text, 
      primary key (tag, id))
      partition by list( tag );
      call set_table_property('holo_reader_basic_part_src', 'orientation', 'column');
      call set_table_property('holo_reader_basic_part_src', 'shard_count', '3');
    commit;
    
    create table holo_reader_basic_part_src_1583161774228 partition of holo_reader_basic_part_src for values in ('foo');
    
    # Ensure that the child tables of the partitioned table have been created and data has been imported.
    postgres=# \d+ holo_reader_basic_part_src
                             Table "public.holo_reader_basic_part_src"
     Column |  Type   | Collation | Nullable | Default | Storage  | Stats target | Description 
    --------+---------+-----------+----------+---------+----------+--------------+-------------
     tag    | text    |           | not null |         | extended |              | 
     id     | integer |           | not null |         | plain    |              | 
     title  | text    |           | not null |         | extended |              | 
     body   | text    |           |          |         | extended |              | 
    Partition key: LIST (tag)
    Indexes:
        "holo_reader_basic_part_src_pkey" PRIMARY KEY, btree (tag, id)
    Partitions: holo_reader_basic_part_src_1583161774228 FOR VALUES IN ('foo')

Reader script parameters

Writer script demo

• Configure a non-partitioned table

  • Configure data generated from MySQL to be imported into a regular Hologres table. The following example shows the configuration for importing data in INSERT mode.

    {
        "type": "job",
        "version": "2.0",
        "steps": [
            {
                "stepType": "mysql",
                "parameter": {
                    "envType": 0,
                    "useSpecialSecret": false,
                    "column": [
                        "<column1>",
                        "<column2>",
                        ......,
                        "<columnN>"
                    ],
                    "tableComment": "",
                    "connection": [
                        {
                            "datasource": "<mysql_source_name>",//MySQL data source name
                            "table": [
                                "<mysql_table_name>"
                            ]
                        }
                    ],
                    "where": "",
                    "splitPk": "",
                    "encoding": "UTF-8"
                },
                "name": "Reader",
                "category": "reader"
            },
            {
                "stepType": "holo",
                "parameter": {
                    "selectedDatabase":"public",
                    "schema": "public",
                    "writeMode": "FIXED_COPY",
                    "maxConnectionCount": 9,
                    "truncate":true,//Cleanup rule
                    "datasource": "<holo_sink_name>",//Hologres data source name
                    "conflictMode": "ignore",
                    "envType": 0,
                    "column": [
                        "<column1>",
                        "<column2>",
                        ......,
                        "<columnN>"
                    ],
                    "tableComment": "",
                    "table": "<holo_table_name>",
                    "reShuffleByDistributionKey":false
                },
                "name": "Writer",
                "category": "writer"
            }
        ],
        "setting": {
            "executeMode": null,
            "errorLimit": {
                "record": "0"
            },
            "locale": "zh_CN",
            "speed": {
                "concurrent": 2,//Job concurrency
                "throttle": false//Throttling
            }
        },
        "order": {
            "hops": [
                {
                    "from": "Reader",
                    "to": "Writer"
                }
            ]
        }
    }

  • The DDL statement for the Hologres table is as follows.

    begin;
    drop table if exists mysql_to_holo_test;
    create table mysql_to_holo_test(
      tag text not null,
      id int not null,
      body text not null,
      brrth date,
      primary key (tag, id));
      call set_table_property('mysql_to_holo_test', 'orientation', 'column');
      call set_table_property('mysql_to_holo_test', 'distribution_key', 'id');
      call set_table_property('mysql_to_holo_test', 'clustering_key', 'birth');
    commit;

• Configure a partitioned table

  Note

  • Hologres supports only LIST partitions. Only a single partition column is supported, and the column must be of the INT4 or TEXT type.

  • Make sure that this parameter matches the partition configuration in the table DDL.

  • Configure the synchronization of data generated from MySQL to child tables of a Hologres partitioned table.

    {
      "type": "job",
      "version": "2.0",
      "steps": [
        {
          "stepType": "mysql",
          "parameter": {
            "envType": 0,
            "useSpecialSecret": false,
            "column": [
              "<column1>",
              "<column2>",
                ......,
              "<columnN>"
            ],
            "tableComment": "",
            "connection": [
              {
                "datasource": "<mysql_source_name>",
                "table": [
                  "<mysql_table_name>"
                ]
              }
            ],
            "where": "",
            "splitPk": "<mysql_pk>",//MySQL primary key field
            "encoding": "UTF-8"
          },
          "name": "Reader",
          "category": "reader"
        },
        {
          "stepType": "holo",
          "parameter": {
            "selectedDatabase": "public",
            "writeMode": "insert",
            "maxConnectionCount": 9,
            "partition": "ds=20201215",//Hologres partition key
            "truncate": "false",
            "datasource": "<holo_sink_name>",//Hologres data source name
            "conflictMode": "ignore",
            "envType": 0,
            "column": [
              "<column1>",
              "<column2>",
                ......,
              "<columnN>"
            ],
            "tableComment": "",
            "table": "<holo_table_name>",
            "reShuffleByDistributionKey":false
          },
          "name": "Writer",
          "category": "writer"
        }
      ],
      "setting": {
        "executeMode": null,
        "failoverEnable": null,
        "errorLimit": {
          "record": "0"
        },
        "speed": {
          "concurrent": 2,//Job concurrency
          "throttle": false//Throttling
        }
      },
      "order": {
        "hops": [
          {
            "from": "Reader",
            "to": "Writer"
          }
        ]
      }
    }

  • The DDL statement of the Hologres table is as follows.

    BEGIN;
    CREATE TABLE public.hologres_parent_table(
      a text ,
      b int,
      c timestamp,
      d text,
      ds text,
      primary key(ds,b)
      )
      PARTITION BY LIST(ds);
    CALL set_table_property('public.hologres_parent_table', 'orientation', 'column');
    CREATE TABLE public.holo_child_1 PARTITION OF public.hologres_parent_table FOR VALUES IN('20201215');
    CREATE TABLE public.holo_child_2 PARTITION OF public.hologres_parent_table FOR VALUES IN('20201216');
    CREATE TABLE public.holo_child_3 PARTITION OF public.hologres_parent_table FOR VALUES IN('20201217');
    COMMIT;

Writer script parameters

Basic parameters

Connection and performance parameters

FIXED_COPY mode-specific parameters

Parameters specific to STAGE mode