DataWorks:Hologres data source

Supported Hologres versions

Hologres V0.7, V0.8, V0.9, V0.10, V1.1, V1.2, and V1.3

Limits

Batch data read and write

• Hologres data sources support only exclusive resource groups for Data Integration.

• Hologres Writer cannot write data to Hologres foreign tables.

• Logic based on which the endpoint of a Hologres data source is obtained during the establishment of a network connection between the Hologres data source and a resource group:

  • If the Hologres data source resides in the same region as the resource group, the endpoint of the Hologres data source is obtained in the following order: classic network endpoint > VPC endpoint > public endpoint.

  • If the Hologres data source resides in a different region from the resource group, the endpoint of the Hologres data source is obtained in the following order: public endpoint > VPC endpoint.

Real-time data write in a database

• You can use only exclusive resource groups for Data Integration to run real-time synchronization tasks.

• A real-time synchronization task cannot be used to synchronize data from a table that has no primary key.

Real-time write of full and incremental data in a single table or a database

When you synchronize data to Hologres, you can write data only to child partitioned tables.

Supported data types

How Hologres Reader and Hologres Writer work

Batch data read and write

Batch data write

Hologres Writer obtains data from a reader and writes data to Hologres based on the conflict processing mode that is configured for the related batch synchronization task.

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 a real-time synchronization task to synchronize data of a single table or synchronize all data in a database

For more information about the configuration procedure, see Configure a real-time synchronization task in DataStudio.

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

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

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 Hologres Reader

• Read data from a non-partitioned Hologres table

  • In the following code, a synchronization task is configured to read data from a non-partitioned Hologres table:

    {
      "type":"job",
      "version":"2.0",// The version number. 
      "steps":[
        {
          "stepType":"holo",// The plug-in name. 
          "parameter":{
            "endpoint": "instance-id-region-endpoint.hologres.aliyuncs.com:port",
            "accessId": "***************", // The AccessKey ID of the account that you use to connect to Hologres. 
            "accessKey": "*******************", // The AccessKey secret of the account that you use to connect to Hologres. 
            "database": "postgres",
            "table": "holo_reader_****",
            "column" : [ // The names of the columns. 
              "tag",
              "id",
              "title"
            ]
          },
          "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"
          }
        ]
      }
    }

  • The following sample DDL statements are used to create a non-partitioned Hologres table:

    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;

• Read data from a partition in a partitioned Hologres table

  • In the following code, a synchronization task is configured to read data from a child partitioned table of a partitioned Hologres table.

    Note

    Exercise caution when you configure the partition parameter.

    {
      "type":"job",
      "version":"2.0",// The version number. 
      "steps":[
        {
          "stepType":"holo",// The plug-in name. 
          "parameter":{
            "endpoint": "instance-id-region-endpoint.hologres.aliyuncs.com:port",
            "accessId": "***************", // The AccessKey ID of the account that you use to connect to Hologres. 
            "accessKey": "*******************", // The AccessKey secret of the account that you use to connect to Hologres. 
            "database": "postgres",
            "table": "holo_reader_basic_****",
            "partition": "tag=foo",
            "column" : [
              "*"
            ],
            "fetchSize": "100"
          },
          "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"
          }
        ]
      }
    }

  • The following sample DDL statements are used to create a partitioned Hologres table:

    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');
    
    # Make sure that the child partitioned table from which you want to read data is created and data is inserted into the child partitioned table. 
    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')

Parameters in code for Hologres Reader

Code for Hologres Writer

• Write data to a non-partitioned table

  • In the following code, a synchronization task is configured to write the data obtained from the memory to a non-partitioned Hologres table in JDBC mode:

    {
        "type": "job",
        "version": "2.0",
        "steps": [
            {
                "stepType": "mysql",
                "parameter": {
                    "envType": 0,
                        "datasource": "<mysql_source_name>",
                    "column": [
                        "<column1>",
                        "<column2>",
                        ......,
                        "<columnN>"
                    ],
                    "connection": [
                        {
                            "datasource": "<mysql_source_name>",// The name of the MySQL data source from which you want to read data.
                            "table": [
                                "<mysql_table_name>"
                            ]
                        }
                    ],
                    "where": "",
                    "splitPk": "",
                    "encoding": "UTF-8"
                },
                "name": "Reader",
                "category": "reader"
            },
            {
                "stepType": "holo",
                "parameter": {
                    "maxConnectionCount": 9,
                    "datasource": "<holo_sink_name>",// The name of the Hologres data source to which you want to write data.
                    "truncate":true,// Specifies whether to delete the existing data in the destination Hologres table before Hologres Writer writes data to the table. 
                    "conflictMode": "ignore",
                    "envType": 0,
                    "column": [
                        "<column1>",
                        "<column2>",
                        ......,
                        "<columnN>"
                    ],
                    "table": "<holo_table_name>"
                },
                "name": "Writer",
                "category": "writer"
            }
        ],
        "setting": {
            "executeMode": null,
            "errorLimit": {
                "record": ""
            },
            "speed": {
                "concurrent": 2,// The maximum number of parallel threads.
                "throttle": false// Specifies whether to enable throttling. The value false indicates that throttling is disabled, and the value true indicates that throttling is enabled.
            }
        },
        "order": {
            "hops": [
                {
                    "from": "Reader",
                    "to": "Writer"
                }
            ]
        }
    }

  • The following sample DDL statements are used to create a non-partitioned Hologres table:

    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;

• Write data to a partitioned table

  Note

  • Hologres supports only list partitioning. You can specify only one column as the partition key column. The data type of the partition key column must be INT4 or TEXT.

  • The value of the partition parameter must match the partition filter expression in the DDL statements that are used to create the destination Hologres table.

  • In the following code, a synchronization task is configured to write the data obtained from the memory to a child partitioned table of a partitioned Hologres table:

    {
      "type": "job",
      "version": "2.0",
      "steps": [
        {
          "stepType": "mysql",
          "parameter": {
            "envType": 0,
            "datasource": "<mysql_source_name>",
            "column": [
              "<column1>",
              "<column2>",
                ......,
              "<columnN>"
            ],
            "connection": [
              {
                "datasource": "<mysql_source_name>",
                "table": [
                  "<mysql_table_name>"
                ]
              }
            ],
            "where": "",
            "splitPk": "<mysql_pk>",// The field that is used for data sharding when MySQL Reader reads data.
            "encoding": "UTF-8"
          },
          "name": "Reader",
          "category": "reader"
        },
        {
          "stepType": "holo",
          "parameter": {
            "maxConnectionCount": 9,
            "partition": "<partition_key>",// The partition key of the destination Hologres table.
            "datasource": "<holo_sink_name>",// The name of the Hologres data source to which you want to write data.
            "conflictMode": "ignore",
            "envType": 0,
            "column": [
              "<column1>",
              "<column2>",
                ......,
              "<columnN>"
            ],
            "table": "<holo_table_name>"
          },
          "name": "Writer",
          "category": "writer"
        }
      ],
      "setting": {
        "executeMode": null,
        "errorLimit": {
          "record": ""
        },
        "speed": {
          "concurrent": 2,// The maximum number of parallel threads.
          "throttle": false// Specifies whether to enable throttling. The value false indicates that throttling is disabled, and the value true indicates that throttling is enabled.
        }
      },
      "order": {
        "hops": [
          {
            "from": "Reader",
            "to": "Writer"
          }
        ]
      }
    }

  • The following sample DDL statements are used to create a partitioned Hologres table:

    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;

Parameters in code for Hologres Writer