DataWorks:Origen de datos de Hologres

Lectura y escritura por lotes

• Hologres Writer no admite escribir datos en tablas externas de Hologres.

• Lógica para obtener el endpoint de Hologres y establecer la conectividad del origen de datos:

  • Para las instancias de Hologres en la región actual, la prioridad de los endpoints es: any Tunnel > Single Tunnel > Public (internet).

  • Para las instancias de Hologres en otras regiones, la prioridad de los endpoints es: Public (internet) > Single Tunnel.

Lectura en tiempo real de una sola tabla

• La versión de Hologres debe ser 2.1 o posterior.

• No se admite la sincronización incremental de tablas particionadas de Hologres.

• No se admite la sincronización de mensajes de cambio DDL en tablas de Hologres.

• La sincronización incremental de Hologres es compatible con a continuación tipos de datos:

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

• Para la sincronización en tiempo real de una sola tabla desde Hologres, habilite Hologres Binlog en la tabla de origen de Hologres. Para obtener más información, consulte Subscribe to Hologres Binlog.

Escritura en tiempo real de bases de datos completas

• Las tareas de sincronización de datos en tiempo real no admiten la sincronización de tablas sin claves primarias.

• Al sincronizar datos en tiempo real desde una base de datos completa de MySQL hacia Hologres, los datos solo se pueden escribir en tablas secundarias de tablas particionadas. No se admite la escritura en la tabla principal de una tabla particionada.

Lectura por lotes

Hologres Reader admite dos modos de lectura:

Modo JDBC (predeterminado)

Lee datos de las tablas de Hologres mediante PSQL. Se inician varias tareas simultáneas según el recuento de fragmentos de la tabla. Cada fragmento corresponde a una tarea Select simultánea:

• Al crear una tabla en Hologres, configure el recuento de fragmentos de la tabla. Para ello, ejecute CALL set_table_property('table_name', 'shard_count', 'xx') dentro de la misma transacción CREATE TABLE.

  De forma predeterminada, la base de datos utiliza el recuento de fragmentos predeterminado, el cual depende de la configuración de la instancia de Hologres.

• La instrucción Select filtra los datos por fragmento mediante la columna integrada hg_shard_id de la tabla.

Modo Arrow (useArrow=true)

Lee datos mediante el protocolo COPY OUT Arrow de Hologres. Este modo admite transferencias comprimidas con LZ4 y ofrece un mejor rendimiento:

• Requiere la versión de Hologres >= 4.0.18.

• Las tareas simultáneas también se dividen según el recuento de fragmentos.

• No admite columnas de función ni columnas constantes. Solo admite columnas físicas de la tabla.

• La transferencia comprimida se habilita automáticamente cuando se cumple el requisito de versión.

Escritura por lotes

Hologres Writer obtiene los datos de protocolo generados por el Reader a través del framework de sincronización de datos. El canal y la estrategia de resolución de conflictos que se utilizan para escribir los datos se determinan mediante las configuraciones writeMode y conflictMode.

Modo de escritura (writeMode)

Recomendaciones de escenarios

Modo de manejo de conflictos (conflictMode)

Configure conflictMode para especificar cómo se procesan los datos recién importados cuando ocurre un conflicto de clave primaria con los datos existentes:

conflictMode solo aplica a las tablas que tienen claves primarias. Para obtener más información sobre el mecanismo de escritura y el rendimiento, consulte Write data to Hologres.

• Cuando conflictMode se establece en Replace (full row update), los datos nuevos sobrescriben los antiguos. Se sobrescriben todas las columnas de la fila. Las columnas que no están incluidas en el mapeo de columnas se establecen forzosamente en NULL.

• Cuando conflictMode se establece en Update, los datos nuevos sobrescriben los antiguos. Solo se sobrescriben las columnas incluidas en el mapeo de columnas.

• Cuando conflictMode se establece en Ignore, los datos nuevos se ignoran.

Sincronización por lotes de una sola tabla

• Orígenes de datos compatibles: todos los tipos de orígenes de datos que admite el módulo Data Integration.

• Guía de configuración: Configure a batch synchronization task in the codeless UI y Configure a batch synchronization task by using the code editor. Para consultar los parámetros completos y los ejemplos de script disponibles en el editor de código, consulte Appendix: Complete parameters and script demos a continuación.

Sincronización en tiempo real de una sola tabla

• Orígenes de datos compatibles: DataHub, Hologres, Kafka y LogHub

• Guía de configuración: Configure a real-time synchronization task.

Sincronización por lotes de bases de datos completas

• Orígenes de datos compatibles: AnalyticDB for MySQL 3.0, ClickHouse, Doris, Hologres, Oracle, PolarDB y SQL Server

• Guía de configuración: Configure a full-database batch synchronization task

Sincronización en tiempo real de bases de datos completas

• Orígenes de datos compatibles: AnalyticDB for OceanBase, MongoDB, MySQL, Oracle, PolarDB, PolarDB-X 2.0 y PostgreSQL

• Guía de configuración: Configure a real-time synchronization task to synchronize data of a single table

Sincronización serverless en tiempo real de bases de datos completas

• Origen de datos compatible: MySQL

• Guía de configuración: Configure a Serverless full-database real-time synchronization task

Configurar una tarea de sincronización por lotes mediante el editor de código

Si desea configurar una tarea de sincronización por lotes mediante el editor de código, defina los parámetros correspondientes en el script según los requisitos unificados de formato de scripts. Para obtener más información, consulte Script mode configuration. A continuación se describen los parámetros que debe configurar para los orígenes de datos al utilizar el editor de código.

Ejemplos de script de Reader

• Configurar una tabla no particionada

  • En el siguiente ejemplo se muestra cómo configurar la lectura de datos desde una tabla no particionada de Hologres hacia la memoria.

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

  • La instrucción DDL de la tabla de Hologres es la siguiente.

    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 una tabla particionada

  • Configure la lectura de datos desde una tabla secundaria de una tabla particionada de Hologres hacia la memoria.

    Nota

    Preste atención a la configuración 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"
          }
        ]
      }
    }

  • La instrucción DDL de la tabla de Hologres es la siguiente.

    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 del script de Reader

Ejemplo de script de Writer

• Configurar una tabla no particionada

  • Configure los datos generados desde MySQL para importarlos a una tabla normal de Hologres. En el siguiente ejemplo se muestra la configuración para importar datos en 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"
                }
            ]
        }
    }

  • La instrucción DDL de la tabla de Hologres es la siguiente.

    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 una tabla particionada

  Nota

  • Hologres solo admite particiones LIST. Únicamente se permite una sola columna de partición, la cual debe ser de tipo INT4 o TEXT.

  • Asegúrese de que este parámetro coincida con la configuración de particiones definida en el DDL de la tabla.

  • Configure la sincronización de los datos generados desde MySQL hacia las tablas secundarias de una tabla particionada de 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"
          }
        ]
      }
    }

  • La instrucción DDL de la tabla de Hologres es la siguiente.

    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 del script Writer

Parámetros básicos

Parámetros de conexión y rendimiento

Parámetros específicos del modo FIXED_COPY

Parámetros específicos del modo STAGE