DataWorks:Source de données Hologres

Lecture et écriture par lots

• Hologres Writer ne prend pas en charge l'écriture de données dans les tables externes Hologres.

• Logique d'obtention de l'endpoint Hologres pour la connectivité de la source de données :

  • Pour les instances Hologres de la région actuelle, la priorité des endpoints est la suivante : any Tunnel > Single Tunnel > Public (internet).

  • Pour les instances Hologres d'autres régions, la priorité des endpoints est la suivante : Public (internet) > Single Tunnel.

Lecture en temps réel d'une seule table

• La version de Hologres doit être 2,1 ou ultérieure.

• La synchronisation incrémentielle des tables partitionnées Hologres n'est pas prise en charge.

• La synchronisation des messages de modification DDL sur les tables Hologres n'est pas prise en charge.

• La synchronisation incrémentielle Hologres prend en charge les types de données suivants :

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

• Pour la synchronisation en temps réel d'une seule table depuis Hologres, vous devez activer Hologres Binlog sur la table source Hologres. Pour plus d'informations, consultez S'abonner à Hologres Binlog.

Écriture en temps réel sur toute la base de données

• Les tâches de synchronisation de données en temps réel ne prennent pas en charge la synchronisation des tables sans clés primaires.

• Lors de la synchronisation des données d'une base de données MySQL complète vers Hologres en temps réel, les données ne peuvent être écrites que dans les tables enfants des tables partitionnées. L'écriture dans la table parent d'une table partitionnée n'est pas prise en charge.

Lecture par lots

Hologres Reader prend en charge deux modes de lecture :

Mode JDBC (par défaut)

Lit les données des tables Hologres via PSQL. Plusieurs tâches concurrentes sont lancées en fonction du nombre de shards de la table, chaque shard correspondant à une tâche Select concurrente :

• Lors de la création d'une table dans Hologres, vous pouvez configurer le nombre de shards de la table en exécutant CALL set_table_property('table_name', 'shard_count', 'xx') au sein de la même transaction CREATE TABLE.

  Par défaut, la base de données utilise le nombre de shards par défaut, qui dépend de la configuration de l'instance Hologres.

• L'instruction Select filtre les données par shard à l'aide de la colonne intégrée hg_shard_id de la table.

Mode Arrow (useArrow=true)

Lit les données via le protocole Hologres COPY OUT Arrow. Ce mode prend en charge le transfert compressé LZ4 et offre des performances supérieures :

• Requiert Hologres version >= 4.0.18.

• Les tâches concurrentes sont également divisées en fonction du nombre de shards.

• Les colonnes de fonction et les colonnes constantes ne sont pas prises en charge. Seules les colonnes physiques de la table sont prises en charge.

• Le transfert compressé est automatiquement activé lorsque la condition de version est remplie.

Écriture par lots

Hologres Writer obtient les données de protocole générées par le Reader via le framework de synchronisation des données. Le canal et la stratégie de résolution des conflits utilisés pour écrire les données sont déterminés par les configurations writeMode et conflictMode.

Mode d'écriture (writeMode)

Recommandations de scénario

Mode de gestion des conflits (conflictMode)

Vous pouvez configurer conflictMode pour spécifier comment les données nouvellement importées sont gérées lorsqu'un conflit de clé primaire se produit avec les données existantes :

conflictMode s'applique uniquement aux tables disposant de clés primaires. Pour plus d'informations sur le mécanisme d'écriture et les performances, consultez Écrire des données dans Hologres.

• Lorsque conflictMode est défini sur Replace (mise à jour complète de la ligne), les nouvelles données écrasent les anciennes données. Toutes les colonnes de la ligne sont écrasées. Les colonnes qui ne sont pas incluses dans le mappage des colonnes sont forcées à NULL.

• Lorsque conflictMode est défini sur Update, les nouvelles données écrasent les anciennes données. Seules les colonnes incluses dans le mappage des colonnes sont écrasées.

• Lorsque conflictMode est défini sur Ignore, les nouvelles données sont ignorées.

Synchronisation par lots d'une seule table

• Sources de données prises en charge : Tous les types de sources de données pris en charge par le module Data Integration.

• Guide de configuration : Configurer une tâche de synchronisation par lots dans l'interface sans code et Configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code. Pour les paramètres complets et les démonstrations de scripts disponibles dans l'éditeur de code, consultez Annexe : Paramètres complets et démonstrations de scripts ci-dessous.

Synchronisation en temps réel d'une seule table

• Sources de données prises en charge : DataHub, Hologres, Kafka et LogHub

• Guide de configuration : Configurer une tâche de synchronisation en temps réel.

Synchronisation par lots sur toute la base de données

• Sources de données prises en charge : AnalyticDB for MySQL 3.0, ClickHouse, Doris, Hologres, Oracle, PolarDB et SQL Server

• Guide de configuration : Configurer une tâche de synchronisation par lots sur toute la base de données

Synchronisation en temps réel sur toute la base de données

• Sources de données prises en charge : AnalyticDB for OceanBase, MongoDB, MySQL, Oracle, PolarDB, PolarDB-X 2.0 et PostgreSQL

• Guide de configuration : Configurer une tâche de synchronisation en temps réel pour synchroniser les données d'une seule table

Temps réel sur toute la base de données en mode serverless

• Source de données prise en charge : MySQL

• Guide de configuration : Configurer une tâche de synchronisation en temps réel sur toute la base de données en mode serverless

Configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code

Si vous souhaitez configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code, vous devez configurer les paramètres associés dans le script selon les exigences de format de script unifié. Pour plus d'informations, consultez Configuration en mode script. Les informations suivantes décrivent les paramètres que vous devez configurer pour les sources de données lors de la configuration d'une tâche de synchronisation par lots à l'aide de l'éditeur de code.

Démonstrations de scripts Reader

• Configurer une table non partitionnée

  • L'exemple suivant montre comment configurer la lecture des données d'une table Hologres non partitionnée vers la mémoire.

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

  • L'instruction DDL de la table Hologres est la suivante.

    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;

• Configurer une table partitionnée

  • Configurez la lecture des données d'une table enfant d'une table partitionnée Hologres vers la mémoire.

    Remarque

    Faites attention à la configuration 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"
          }
        ]
      }
    }

  • L'instruction DDL de la table Hologres est la suivante.

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

Paramètres du script Reader

Démonstration de script Writer

• Configurer une table non partitionnée

  • Configurez l'importation des données générées depuis MySQL dans une table Hologres régulière. L'exemple suivant montre la configuration pour l'importation de données en mode 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"
                }
            ]
        }
    }

  • L'instruction DDL de la table Hologres est la suivante.

    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;

• Configurer une table partitionnée

  Remarque

  • Hologres prend uniquement en charge les partitions LIST. Une seule colonne de partition est prise en charge, et la colonne doit être de type INT4 ou TEXT.

  • Assurez-vous que ce paramètre correspond à la configuration de partition dans la DDL de la table.

  • Configurez la synchronisation des données générées depuis MySQL vers les tables enfants d'une table partitionnée 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"
          }
        ]
      }
    }

  • L'instruction DDL de la table Hologres est la suivante.

    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;

Paramètres du script Writer

Paramètres de base

Paramètres de connexion et de performance

Paramètres spécifiques au mode FIXED_COPY

Paramètres spécifiques au mode STAGE