DataWorks:Fonte de dados Hologres

Leitura e escrita em lote

• O Hologres Writer não permite gravar dados em tabelas externas do Hologres.

• Lógica para obter o endpoint do Hologres visando a conectividade da source 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 2,1 ou superior.

• Não há suporte para sincronização incremental de tabelas particionadas do Hologres.

• Não há suporte para sincronização de mensagens de alteração DDL em tabelas do Hologres.

• 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 sincronizar uma única tabela em tempo real a partir do Hologres, ative o Binlog do Hologres na tabela de origem. Para mais informações, consulte Assinar o Binlog do Hologres.

Escrita em tempo real de banco de dados completo

• Tarefas de sincronização de dados em tempo real não permitem sincronizar tabelas sem chaves primárias.

• Ao sincronizar dados de um banco de dados MySQL completo para o Hologres em tempo real, a gravação ocorre apenas em tabelas filhas de tabelas particionadas. Não há suporte para gravação na tabela pai de uma tabela particionada.

Leitura em lote

O Hologres Reader oferece dois modos de leitura:

Modo JDBC (padrão)

Lê dados das tabelas do Hologres por meio do PSQL. Várias tarefas concorrentes são iniciadas com base na contagem de shards da tabela, onde cada shard corresponde a uma tarefa Select concorrente:

• Ao crie uma tabela no Hologres, configure a contagem de shards executando CALL set_table_property('table_name', 'shard_count', 'xx') dentro da mesma transação CREATE TABLE.

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

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

Modo Arrow (useArrow=true)

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

• Requer versão do Hologres >= 4.0.18.

• As tarefas concorrentes também são divididas com base na contagem de shards.

• Não há suporte para colunas de função e colunas constantes. Apenas colunas físicas da tabela são aceitas.

• A transferência compactada é ativada automaticamente quando o requisito de versão é atendido.

Escrita em lote

O Hologres Writer obtém os dados de protocolo gerados pelo Reader por meio do framework de sincronização de dados. O canal e a estratégia de resolução de conflitos usados para gravar os dados são determinados pelas configurações writeMode e conflictMode.

Modo de escrita (writeMode)

Recomendações de cenário

Modo de tratamento de conflitos (conflictMode)

Configure conflictMode para especifique como tratar os dados recém-importados quando ocorre um conflito de chave primária com dados existentes:

O parâmetro conflictMode aplica-se apenas a tabelas com chaves primárias. Para mais informações sobre o mecanismo de escrita e desempenho, consulte Escrever dados no Hologres.

• Se conflictMode estiver defina como Replace (full row update), os novos dados substituirão os antigos. Todas as colunas da linha serão sobrescritas. As colunas não incluídas no mapeamento de colunas serão forçadas para NULL.

• Se conflictMode estiver defina como Update, os novos dados substituirão os antigos. Apenas as colunas incluídas no mapeamento de colunas serão sobrescritas.

• Se conflictMode estiver defina como Ignore, os novos dados serão ignorados.

Sincronização em lote de tabela única

• Fontes de dados compatíveis: Todos os tipos de fontes de dados compatíveis com o módulo Data Integration.

• Guia de configuração: configure uma tarefa de sincronização em lote na interface visual e configure uma tarefa de sincronização em lote usando o editor de código. Para obter a lista completa de parâmetros e exemplos de scripts disponíveis no editor de código, consulte o Apêndice: Parâmetros completos e exemplos de scripts abaixo.

Sincronização em tempo real de tabela única

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

• Guia de configuração: configure uma tarefa de sincronização em tempo real.

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 uma tarefa de sincronização em lote de banco de dados completo

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 uma tarefa de sincronização em tempo real para sincronizar dados de uma única tabela

Tempo real de banco de dados completo Serverless

• Fonte de dados compatível: MySQL

• Guia de configuração: configure uma tarefa de sincronização em tempo real de banco de dados completo Serverless

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

Para configure uma tarefa de sincronização em lote usando o editor de código, defina os parâmetros relevantes no script conforme os requisitos unificados de formato de script. Para mais informações, consulte Configuração em modo de script. As informações a seguir descrevem os parâmetros necessários para as fontes de dados ao utilizar o editor de código.

Exemplos de script do Reader

• configure uma tabela não particionada

  • O exemplo a seguir mostra como configure 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;

• configure uma tabela particionada

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

    Nota

    Atente-se à 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 do script do Reader

Exemplo de script do Writer

• configure 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 para a 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;

• configure uma tabela particionada

  Nota

  • O Hologres oferece suporte apenas a partições LIST. Apenas uma única coluna de partição é compatível, e a coluna 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 do 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