大数据开发治理平台 DataWorks：Hologres数据源

配置非分区表

• 配置从Hologres非分区表读取数据至内存，如下所示。

  {
    "type":"job",
    "version":"2.0",//版本号。
    "steps":[
      {
        "stepType":"holo",//插件名。
        "parameter":{
          "endpoint": "instance-id-region-endpoint.hologres.aliyuncs.com:port",
          "accessId": "***************", //访问Hologres的accessId。
          "accessKey": "*******************", //访问Hologres的accessKey。
          "database": "postgres",
          "table": "holo_reader_****",
          "column" : [ //字段。
            "tag",
            "id",
            "title"
          ]
        },
        "name":"Reader",
        "category":"reader"
      },
      {
        "stepType":"stream",
        "parameter":{},
        "name":"Writer",
        "category":"writer"
      }
    ],
    "setting":{
      "errorLimit":{
        "record":"0"//错误记录数。
      },
      "speed":{
        "throttle":true,//当throttle值为false时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
        "concurrent":1,//作业并发数。 
                    "mbps":"12"//限流，此处1mbps = 1MB/s。
      }
    },
    "order":{
      "hops":[
        {
          "from":"Reader",
          "to":"Writer"
        }
      ]
    }
  }

• Hologres表的DDL语句，如下所示。

  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;

配置分区表

• 配置从内存产生的数据同步至Hologres分区表的子表。

  说明

  请注意partition的配置。

  {
    "type":"job",
    "version":"2.0",//版本号。
    "steps":[
      {
        "stepType":"holo",//插件名。
        "parameter":{
          "endpoint": "instance-id-region-endpoint.hologres.aliyuncs.com:port",
          "accessId": "***************", //访问Hologres的accessId。
          "accessKey": "*******************", //访问Hologres的accessKey。
          "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"//错误记录数。
      },
      "speed":{
        "throttle":true,//当throttle值为false时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
        "concurrent":1,//作业并发数。
                    "mbps":"12"//限流，此处1mbps = 1MB/s。
      }
    },
    "order":{
      "hops":[
        {
          "from":"Reader",
          "to":"Writer"
        }
      ]
    }
  }

• Hologres表的DDL语句，如下所示。

  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');
  
  # 确保分区表子表已经创建且导入数据。
  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')

配置非分区表

• 配置从内存产生的数据导入至Hologres普通表，示例为通过JDBC模式导入的配置。

  {
      "type": "job",
      "version": "2.0",
      "steps": [
          {
              "stepType": "mysql",
              "parameter": {
                  "envType": 0,
                      "datasource": "<mysql_source_name>",
                  "column": [
                      "<column1>",
                      "<column2>",
                      ......,
                      "<columnN>"
                  ],
                  "connection": [
                      {
                          "datasource": "<mysql_source_name>",//mysql数据源名
                          "table": [
                              "<mysql_table_name>"
                          ]
                      }
                  ],
                  "where": "",
                  "splitPk": "",
                  "encoding": "UTF-8"
              },
              "name": "Reader",
              "category": "reader"
          },
          {
              "stepType": "holo",
              "parameter": {
                  "maxConnectionCount": 9,
                  "datasource": "<holo_sink_name>",//Hologres数据源名称
                  "truncate":true,//清理规则。
                  "conflictMode": "ignore",
                  "envType": 0,
                  "column": [
                      "<column1>",
                      "<column2>",
                      ......,
                      "<columnN>"
                  ],
                  "table": "<holo_table_name>"
              },
              "name": "Writer",
              "category": "writer"
          }
      ],
      "setting": {
          "executeMode": null,
          "errorLimit": {
              "record": ""
          },
          "speed": {
              "concurrent": 2,//作业并发数
              "throttle": false//限流
          }
      },
      "order": {
          "hops": [
              {
                  "from": "Reader",
                  "to": "Writer"
              }
          ]
      }
  }

• Hologres表的DDL语句，如下所示。

  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;

配置分区表

• 配置从内存产生的数据同步至Hologres分区表的子表。

  {
    "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>",//mysql的pk字段
          "encoding": "UTF-8"
        },
        "name": "Reader",
        "category": "reader"
      },
      {
        "stepType": "holo",
        "parameter": {
          "maxConnectionCount": 9,
          "partition": "<partition_key>",//Hologres分区键
          "datasource": "<holo_sink_name>",//Hologres数据源名
          "conflictMode": "ignore",
          "envType": 0,
          "column": [
            "<column1>",
            "<column2>",
              ......,
            "<columnN>"
          ],
          "table": "<holo_table_name>"
        },
        "name": "Writer",
        "category": "writer"
      }
    ],
    "setting": {
      "executeMode": null,
      "errorLimit": {
        "record": ""
      },
      "speed": {
        "concurrent": 2,//作业并发数
        "throttle": false//限流
      }
    },
    "order": {
      "hops": [
        {
          "from": "Reader",
          "to": "Writer"
        }
      ]
    }
  }

• Hologres表的DDL语句，如下所示。

  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;