DataWorks:Synchronisation en temps réel d'une table Hologres vers Kafka

1. Sélectionner un type de tâche de synchronisation

1. Connectez-vous à la console DataWorks. Dans la région cible, cliquez sur Data Integration > Data Integration dans le volet de navigation de gauche. Sélectionnez un espace de travail dans la liste déroulante et cliquez sur Go to Data Integration.

2. Dans le volet de navigation de gauche, cliquez sur Sync Tasks. En haut de la page, cliquez sur Create synchronization task et configurez les paramètres suivants.

   • Data Source And Destination : Hologres→Kafka

   • New Task Name : Personnalisez un nom pour la tâche de synchronisation.

   • Synchronization Mode : single-table real-time.

   • Étape Synchronization Mode : Sélectionnez full synchronization.

2. Configurer le réseau et les ressources

1. Dans la section Configure Network Connections and Resource Groups, sélectionnez le Resource Group pour la tâche de synchronisation. Vous pouvez également définir l'Task Resource Usage en UC (Unités de calcul).

2. Pour Source Data Source, sélectionnez une source de données Hologres. Pour Destination data source, sélectionnez une source de données Kafka. Ensuite, cliquez sur Test Connection.

3. Une fois les tests de connectivité pour les sources de données source et destination réussis, cliquez sur Next step.

3. Configurer le lien de synchronisation

4. Alertes

Pour éviter que des erreurs de tâche ne provoquent des retards dans la synchronisation des données métier, définissez une politique d'alerte pour la tâche de synchronisation.

1. Cliquez sur Alert Settings dans le coin supérieur droit de la page pour ouvrir la page de configuration des Alert Rule Configurations for Real-time Synchronization Subnode.

2. Cliquez sur Add Alert Rule pour configurer une règle d'alerte.

   Remarque

   Les règles d'alerte que vous définissez ici s'appliquent aux sous-tâches de synchronisation en temps réel générées par cette tâche. Après avoir configuré la tâche, vous pouvez afficher et modifier les règles d'alerte pour ces sous-tâches sur la page Exécuter et gérer les tâches de synchronisation en temps réel.

3. Gérez les règles d'alerte.

   Pour les règles d'alerte existantes, utilisez le commutateur pour les activer ou les désactiver. Vous pouvez également envoyer des alertes à différents destinataires en fonction du niveau d'alerte.

5. Paramètres avancés

La tâche de synchronisation propose plusieurs paramètres que vous pouvez modifier selon vos besoins.

1. Cliquez sur advanced settings dans le coin supérieur droit de la page pour ouvrir la page des paramètres avancés.

2. Sur la page advanced settings, modifiez les valeurs des paramètres selon vos besoins.

6. Groupe de ressources

Cliquez sur Configure Resource Group dans le coin supérieur droit pour afficher et changer le groupe de ressources actuel de la tâche.

7. Exécuter la tâche de synchronisation

1. Après avoir terminé tous les paramètres, cliquez sur Complete en bas de la page.

2. Sur la page Data Integration > Synchronization Task, recherchez la tâche que vous avez créée et cliquez sur Start dans la colonne Operations.

3. Cliquez sur le Name/ID de la tâche correspondante dans la Task List pour afficher son processus d'exécution détaillé.

Afficher l'état de la tâche de synchronisation

Après la création de la tâche de synchronisation, accédez à la page Tasks pour afficher toutes les tâches de synchronisation de l'espace de travail et leurs informations de base.

La liste des tâches affiche les colonnes suivantes : Name/ID, Data Source Synchronization Solution (type de synchronisation, source et destination), Status, Execution Overview, resource group et Actions.

• Vous pouvez Start ou Stop une tâche de synchronisation dans la colonne Actions. Sous More, vous pouvez effectuer d'autres opérations, telles que Edit et View.

• Pour les tâches démarrées, vous pouvez afficher l'état d'exécution de base dans l'Execution Overview et cliquer sur la zone de résumé correspondante pour afficher les détails d'exécution.

Une tâche de synchronisation en temps réel d'une table Hologres vers Kafka comprend les trois étapes suivantes :

• Structure Migration : inclut la méthode de création de la table de destination (table existante ou création automatique de table). Si vous sélectionnez la création automatique de table, l'instruction DDL (Data Definition Language) de création de la table est affichée.

• Full initialization : Si vous sélectionnez Full Sync pour l'étape de synchronisation de votre tâche, la progression de l'initialisation complète est affichée ici.

• Real-time Data Synchronization : Contient des statistiques sur la synchronisation en temps réel, y compris le trafic de lecture/écriture en temps réel, les données erronées, les basculements et les journaux d'opérations.

Réexécuter une tâche

Dans des cas particuliers, par exemple lorsque vous devez modifier les champs synchronisés ou ajuster les informations de la table cible, cliquez sur Rerun dans la colonne Operations de la tâche de synchronisation. Cette action synchronise les champs ajustés et autres modifications vers la cible. Le processus ignore les tables inchangées précédemment synchronisées.

• Pour exécuter à nouveau la tâche sans aucune modification, cliquez sur Rerun.

• Si vous modifiez la tâche, cliquez sur Complete après avoir effectué vos modifications. L'action de la tâche devient Apply Updates. Cliquer sur Apply Updates réexécute la tâche avec la nouvelle configuration.

Canal CDC

Canal CDC est un format de données CDC défini par Alibaba Canal.

• Champs et significations

  Nom du champ	Signification de la valeur du champ
  id	La valeur de ce champ est fixée à 0.
  database	Le nom de la base de données Hologres.
  table	Le nom de la table Hologres.
  pkNames	La colonne de clé primaire de la table Hologres.
  isDdl	Indique si les journaux binaires enregistrent les modifications DDL. La valeur de ce champ est fixée à false car la synchronisation des messages pour les modifications DDL sur la table Hologres n'est pas prise en charge.
  type	Le type de modification DML. Valeurs valides : INSERT, UPDATE et DELETE. Remarque Une opération de modification sur la table Hologres génère deux enregistrements dont le type est UPDATE. Les enregistrements seront écrits dans le topic Kafka. L'un des enregistrements correspond au contenu des données avant la modification. L'autre correspond au contenu des données après la modification. Si vous souhaitez synchroniser toutes les données de la table Hologres vers le topic Kafka, la valeur du champ type est fixée à INSERT.
  es	Un horodatage de 13 chiffres en millisecondes. L'horodatage indique l'heure à laquelle les données de la table Hologres sont modifiées. Si vous souhaitez synchroniser toutes les données de la table Hologres vers le topic Kafka, la valeur du champ es est fixée à 0.
  ts	Un horodatage de 13 chiffres en millisecondes. L'horodatage indique l'heure à laquelle les journaux binaires générés pour la table Hologres sont lus par la tâche de synchronisation.
  sql	Le code SQL qui enregistre les modifications DDL lorsque les journaux binaires générés pour la table Hologres contiennent les modifications DDL. La valeur de ce champ est fixée à une chaîne vide car la synchronisation des messages pour les modifications DDL sur la table Hologres n'est pas prise en charge.
  sqlType	Les types de données de champ SQL auxquels correspondent les types de données des champs de la table Hologres. Correspondances entre les types de données Hologres et les valeurs valides de sqlType : bigint : -5 decimal dont l'échelle n'est pas 0 : 3 decimal dont l'échelle est 0 : -5 boolean : 16 date : 91 float4 : 6 float8 : 8 integer : 4 smallint : 5 json : 12 text : 12 varchar : 12 timestamp : 93 timestamptz : 93 bigserial : -5 bytea : 12 char : 12 serial : 4 time : 92 int4[] : 12 int8[] : 12 float4[] : 12 float8[] : 12 boolean[] : 12 text[] : 12
  mysqlType	Les types de données de champ MySQL auxquels correspondent les types de données des champs de la table Hologres. Correspondances entre les types de données Hologres et les valeurs valides de mysqlType : bigint : BIGINT int4 : INT decimal dont l'échelle n'est pas 0 : DECIMAL(xx,xx) decimal dont l'échelle est 0 : BIGINT boolean : BOOLEAN date : DATE float4 : FLOAT float8 : DOUBLE integer : INT smallint : SMALLINT json : TEXT text : TEXT varchar : VARCHAR(xx) timestamp : DATETIME(6) timestamptz : DATETIME(6) bigserial : BIGINT bytea : TEXT char : TEXT serial : INT time : TIME(6) int4[] : TEXT int8[] : TEXT float4[] : TEXT float8[] : TEXT boolean[] : TEXT text[] : TEXT
  data	Les modifications de données dans la table Hologres. Les noms de champs de la table Hologres sont utilisés comme clés, et les modifications de données dans les champs sont sérialisées en chaînes et utilisées comme valeurs. Ensuite, les clés et les valeurs sont organisées sous forme de chaînes au format JSON. Pour plus d'informations sur la sérialisation, consultez la Description de la sérialisation JSON.
  old	Deux enregistrements dont le type est UPDATE sont générés pour une opération de modification sur la table Hologres et écrits dans le topic Kafka. Les enregistrements correspondent au contenu des données avant et après la modification. Dans le premier enregistrement, le champ old est utilisé pour enregistrer le contenu des données dans la table Hologres avant la modification. Le champ data est utilisé pour enregistrer le contenu des données pour les modifications DML autres que UPDATE.

• Exemple de données au format Canal JSON correspondant aux modifications de données générées par une opération INSERT dans les journaux binaires de la table Hologres

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "INSERT",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Exemple de données au format Canal JSON correspondant à toutes les données synchronisées depuis la table Hologres

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "INSERT",
      "es": 0,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Exemple de deux enregistrements de données au format Canal JSON correspondant aux modifications de données générées par une opération UPDATE dans les journaux binaires de la table Hologres

  // Data content before a change
  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "UPDATE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "old": [
          {
              "bigint": "0",
              "integer": "0",
              "smallint": "0"
          }
      ],
      "data": null
  }
  // Data content after a change
  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "UPDATE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Exemple de données au format Canal JSON correspondant aux modifications de données générées par une opération DELETE dans les journaux binaires de la table Hologres

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "DELETE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

Json

Le format JSON utilise les noms de champs des journaux binaires Hologres comme clés et sérialise les valeurs de champ en chaînes. Les paires clé-valeur résultantes sont écrites dans le topic Kafka sous forme de chaînes au format JSON.