DataWorks：Hologres資料來源

離線讀寫

• Hologres Writer不支援將資料寫入Hologres的外部表格。

• Hologres資料來源連通性擷取Hologres端點的邏輯：

  • 當前地區的Hologres執行個體，Hologres端點擷取順序：any Tunnel > Single Tunnel > Public（公網）。

  • 跨地區的Hologres執行個體，Hologres端點擷取順序：Public（公網）> Single Tunnel。

單表即時讀

• Hologres版本必須在2.1以上。

• 不支援Hologres分區表的增量同步處理。

• 不支援Hologres表DDL變更訊息同步。

• Hologres增量同步處理支援的資料類型包括以下類型：

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

• Hologres單表即時同步時，需開啟源端的Hologres資料庫的表Hologres Binlog，詳情可參見訂閱Hologres Binlog。

整庫即時寫

• 即時資料同步任務暫不支援同步沒有主鍵的表。

• MySQL整庫即時同步資料至Hologres時，目前僅支援將資料寫入分區表子表，暫不支援寫入資料至分區表父表。

離線讀

Hologres Reader支援兩種讀模數式：

JDBC模式（預設）

通過PSQL讀取Hologres表中的資料，根據表的Shard Count發起多個並發，每個Shard對應一個Select並發任務：

• Hologres在建立表時，在同一個CREATE TABLE事務中，通過CALL set_table_property('table_name', 'shard_count', 'xx')配置表的Shard Count。

  預設情況下，使用資料庫預設的Shard Count，具體數值取決於Hologres執行個體的配置。

• Select語句通過表的內建列hg_shard_id的Shard篩選資料。

Arrow模式（useArrow=true）

通過Hologres的COPY OUT Arrow協議讀取資料，支援LZ4壓縮傳輸，效能更高：

• 需要Hologres版本 >= 4.0.18。

• 同樣按照Shard Count拆分並發任務。

• 不支援函數列和常量列（只支援表中的物理列）。

• 自動啟用壓縮傳輸（版本滿足時）。

離線寫

Hologres Writer通過資料同步架構擷取Reader產生的協議資料，根據writeMode（寫入模式）和conflictMode（衝突策略）的配置決定寫入資料時的通道和衝突解決方案策略。

寫入模式（writeMode）

情境推薦

衝突處理模式（conflictMode）

您可以通過配置conflictMode，決定新匯入的資料和已有資料的主鍵發生衝突時，如何處理新匯入的資料：

• conflictMode為Replace（整行更新）模式時，新資料覆蓋舊資料，整行所有列全部覆蓋，沒有配置列映射的欄位會強制寫入NULL。

• conflictMode為Update（更新）模式時，新資料覆蓋舊資料，只覆蓋配置有列映射的欄位。

• conflictMode為Ignore（忽略）模式時，忽略新資料。

單表離線

• 支援資料來源：Data Integration模組資料來源支援的所有資料來源類型。

• 配置指導：嚮導模式配置、指令碼模式配置。指令碼模式配置的全量參數和指令碼Demo請參見下文的附錄：指令碼Demo與參數說明。

單表即時

• 支援資料來源：DataHub、Hologres、Kafka、LogHub

• 配置指導：單表即時同步任務配置。

整庫離線

• 支援資料來源：AnalyticDB for MySQL 3.0、ClickHouse、Doris、Hologres、Oracle、PolarDB、SQL Server

• 配置指導：整庫離線同步任務配置

整庫即時

• 支援資料來源：AnalyticDB for OceanBase、MongoDB、MySQL、Oracle、PolarDB、PolarDB-X 2.0、PostgreSQL

• 配置指導：整庫即時同步任務配置

Serverless整庫即時

• 支援資料來源：MySQL

• 配置指導：Serverless同步任務配置

離線任務指令碼配置方式

如果您配置離線任務時使用指令碼模式的方式進行配置，您需要按照統一的指令碼格式要求，在任務指令碼中編寫相應的參數，詳情請參見指令碼模式配置，以下為您介紹指令碼模式下資料來源的參數配置詳情。

Reader指令碼Demo

• 配置非分區表

  • 配置從Hologres非分區表讀取資料至記憶體，如下所示。

    {
        "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"
                }
            ]
        }
    }

  • 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的配置。

    {
        "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"
          }
        ]
      }
    }

  • 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')

Reader指令碼參數

Writer指令碼Demo

• 配置非分區表

  • 配置從MySQL產生的資料匯入至Hologres普通表，樣本為通過INSERT模式匯入的配置。

    {
        "type": "job",
        "version": "2.0",
        "steps": [
            {
                "stepType": "mysql",
                "parameter": {
                    "envType": 0,
                    "useSpecialSecret": false,
                    "column": [
                        "<column1>",
                        "<column2>",
                        ......,
                        "<columnN>"
                    ],
                    "tableComment": "",
                    "connection": [
                        {
                            "datasource": "<mysql_source_name>",//mysql資料來源名
                            "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,//清理規則
                    "datasource": "<holo_sink_name>",//Hologres資料來源名稱
                    "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,//作業並發數
                "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僅支援LIST分區，分區Column僅支援單個Column分區，且僅支援INT4或TEXT類型。

  • 請確認該參數和表DDL的分區配置匹配。

  • 配置從MySQL產生的資料同步至Hologres分區表的子表。

    {
      "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的pk欄位
            "encoding": "UTF-8"
          },
          "name": "Reader",
          "category": "reader"
        },
        {
          "stepType": "holo",
          "parameter": {
            "selectedDatabase": "public",
            "writeMode": "insert",
            "maxConnectionCount": 9,
            "partition": "ds=20201215",//Hologres分區鍵
            "truncate": "false",
            "datasource": "<holo_sink_name>",//Hologres資料來源名
            "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,//作業並發數
          "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;

Writer指令碼參數

基礎參數

串連與績效參數

FIXED_COPY模式特有參數

STAGE模式特有參數