DataWorks:Fonte de dados do Hologres

Leitura e gravação em lote

• O Hologres Writer não oferece suporte à gravação de dados em tabelas externas do Hologres.

• Lógica de obtenção do endpoint do Hologres para conectividade da fonte de dados:

  • Para instâncias do Hologres na região atual, a prioridade do endpoint é: any Tunnel > Single Tunnel > Public (internet).

  • Para instâncias do Hologres em outras regiões, a prioridade do endpoint é: Public (internet) > Single Tunnel.

Leitura em tempo real de tabela única

• A versão do Hologres deve ser a 2.1 ou superior.

• O Hologres não oferece suporte à sincronização incremental de tabelas particionadas.

• O Hologres não oferece suporte à sincronização de mensagens de alteração de DDL em suas tabelas.

• A sincronização incremental do Hologres aceita os seguintes tipos de dados:

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

• Para a sincronização em tempo real de tabela única a partir do Hologres, ative o Binlog do Hologres na tabela de origem. Para obter mais informações, consulte Subscribe to Hologres Binlog.

Gravação em tempo real de banco de dados completo

• As tarefas de sincronização de dados em tempo real não oferecem suporte à sincronização de tabelas sem chaves primárias.

• Ao sincronizar dados de um banco de dados completo do MySQL para o Hologres em tempo real, grave os dados apenas em tabelas filhas de tabelas particionadas. O sistema não oferece suporte à gravação na tabela pai de uma tabela particionada.

Leitura em lote

O Hologres Reader aceita dois modos de leitura:

Modo JDBC (padrão)

Lê dados de tabelas do Hologres por meio do PSQL. O sistema inicia várias tarefas simultâneas com base na contagem de shards da tabela. Cada shard corresponde a uma tarefa Select simultânea:

• É necessário que você crie uma tabela no Hologres e configure a contagem de shards da tabela executando CALL set_table_property('table_name', 'shard_count', 'xx') dentro da mesma transação CREATE TABLE.

  Por padrão, o banco de dados usa a contagem de shards padrão, que depende da configuração da instância do Hologres.

• A instrução Select filtra dados por shard usando a coluna interna hg_shard_id da tabela.

Modo Arrow (useArrow=true)

Lê dados por meio do protocolo Hologres COPY OUT Arrow. Este modo aceita transferência compactada LZ4 e oferece maior desempenho:

• atualize para a versão do Hologres >= 4.0.18.

• O sistema também divide as tarefas simultâneas com base na contagem de shards.

• O modo não aceita colunas de função nem colunas constantes. Apenas colunas físicas na tabela são compatíveis.

• A transferência compactada é ativada automaticamente quando a versão atende ao requisito.

Gravação em lote

O Hologres Writer obtém os dados de protocolo gerados pelo Reader por meio da estrutura de sincronização de dados. As configurações writeMode e conflictMode determinam o canal e a estratégia de resolução de conflitos usados para gravar os dados.

Modo de gravação (writeMode)

Recomendações de cenário

Modo de tratamento de conflitos (conflictMode)

Configure o conflictMode para especificar como o sistema trata os dados recém-importados ao ocorrer um conflito de chave primária com os dados existentes:

O conflictMode se aplica apenas a tabelas com chaves primárias. Para obter mais informações sobre o mecanismo de gravação e o desempenho, consulte Write data to Hologres.

• Quando você define o conflictMode como Replace (full row update), os novos dados substituem os antigos. O sistema substitui todas as colunas na linha e força as colunas não incluídas no mapeamento de colunas como NULL.

• Quando você define o conflictMode como Update, os novos dados substituem os antigos. O sistema substitui apenas as colunas incluídas no mapeamento.

• Quando você define o conflictMode como Ignore, o sistema ignora os novos dados.

Sincronização em lote de tabela única

• Fontes de dados compatíveis: todos os tipos de fontes de dados com suporte no módulo Data Integration.

• Guia de configuração: Configure a batch synchronization task in the codeless UI e Configure a batch synchronization task by using the code editor. Para obter a lista completa de parâmetros e exemplos de script disponíveis no editor de código, consulte Appendix: Complete parameters and script demos abaixo.

Sincronização em tempo real de tabela única

• Fontes de dados compatíveis: DataHub, Hologres, Kafka e LogHub

• Guia de configuração: Configure a real-time synchronization task.

Sincronização em lote de banco de dados completo

• Fontes de dados compatíveis: AnalyticDB for MySQL 3.0, ClickHouse, Doris, Hologres, Oracle, PolarDB e SQL Server

• Guia de configuração: Configure a full-database batch synchronization task

Sincronização em tempo real de banco de dados completo

• Fontes de dados compatíveis: AnalyticDB for OceanBase, MongoDB, MySQL, Oracle, PolarDB, PolarDB-X 2.0 e PostgreSQL

• Guia de configuração: Configure a real-time synchronization task to synchronize data of a single table

Tempo real de banco de dados completo Serverless

• Fonte de dados compatível: MySQL

• Guia de configuração: Configure a Serverless full-database real-time synchronization task

Configurar uma tarefa de sincronização em lote usando o editor de código

Para configurar uma tarefa de sincronização em lote no editor de código, configure e defina os parâmetros relacionados no script de acordo com os requisitos unificados de formato. Para obter mais informações, consulte Script mode configuration. As seções a seguir descrevem os parâmetros obrigatórios para fontes de dados ao configurar uma tarefa de sincronização em lote no editor de código.

Exemplos de script do Reader

• Configurar uma tabela não particionada

  • O exemplo a seguir mostra como configurar a leitura de dados de uma tabela não particionada do Hologres para a memória.

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

  • A instrução DDL da tabela do Hologres é a seguinte:

    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;

• Configurar uma tabela particionada

  • Configure a leitura de dados de uma tabela filha de uma tabela particionada do Hologres para a memória.

    Nota

    Preste atenção na configuração de 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"
          }
        ]
      }
    }

  • A instrução DDL da tabela do Hologres é a seguinte:

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

Parâmetros de script do Reader

Exemplos de script do Writer

• Configurar uma tabela não particionada

  • Configure a importação de dados gerados no MySQL para uma tabela regular do Hologres. O exemplo a seguir mostra a configuração para importar dados no modo 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 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"
                }
            ]
        }
    }

  • A instrução DDL da tabela do Hologres é a seguinte:

    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;

• Configurar uma tabela particionada

  Nota

  • O Hologres aceita apenas partições LIST. Apenas uma única coluna de partição é compatível, e ela deve ser do tipo INT4 ou TEXT.

  • Certifique-se de que este parâmetro corresponda à configuração de partição no DDL da tabela.

  • Configure a sincronização de dados gerados no MySQL para tabelas filhas de uma tabela particionada do 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 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"
          }
        ]
      }
    }

  • A instrução DDL da tabela do Hologres é a seguinte:

    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;

Parâmetros de script do Writer

Parâmetros básicos

Parâmetros de conexão e desempenho

Parâmetros específicos do modo FIXED_COPY

Parâmetros específicos do modo STAGE