ApsaraDB for OceanBase (Deprecated):Synchroniser les données d'OceanBase Database vers une instance RocketMQ

Contexte

Message Queue for Apache RocketMQ est un middleware distribué orienté messages, conçu par Alibaba Cloud sur la base d'Apache RocketMQ. Il se distingue par sa faible latence, sa capacité à gérer une concurrence élevée et sa grande fiabilité. La fonctionnalité de synchronisation des données du service de transmission des données vous permet de synchroniser en temps réel les données entre une table physique d'OceanBase Database et une source de données RocketMQ, étendant ainsi les capacités de traitement des messages.

Pour plus d'informations sur les formats de données utilisés pour les deux types de locataires, consultez la section Formats de données utilisés lors de la transmission des données d'OceanBase Database vers un système de file d'attente de messages.

Prérequis

• Le service de transmission des données dispose des privilèges nécessaires pour accéder aux ressources cloud. Pour plus d'informations, consultez la section Accorder des privilèges aux rôles pour la transmission des données.

• Créez un utilisateur de base de données dédié pour la synchronisation des données dans la base OceanBase source et attribuez-lui les privilèges appropriés. Pour plus d'informations, consultez la section Créer un utilisateur de base de données.

Limites

• Le service de transmission des données prend en charge les instances RocketMQ versions 4.x et 5.x, y compris les éditions commerciales et communautaires.

• Seules les tables physiques peuvent être synchronisées.

• Lors de la synchronisation des données, le service de transmission des données autorise la suppression d'une table avant la création d'une nouvelle. Autrement dit, vous pouvez exécuter DROP TABLE puis CREATE TABLE. En revanche, il n'est pas possible de créer une nouvelle table en renommant une table existante. Vous ne pouvez donc pas exécuter RENAME TABLE a TO a_tmp.

• Le nom d'une table à synchroniser, ainsi que les noms des colonnes de cette table, ne doivent pas contenir de caractères chinois.

• Le service de transmission des données prend en charge la migration d'un objet uniquement si les conditions suivantes sont remplies : le nom de la base de données, le nom de la table et le nom de la colonne de l'objet sont encodés en ASCII et ne contiennent aucun caractère spécial. Les caractères spéciaux incluent les sauts de ligne, les espaces et les caractères suivants : . | " ' ` ( ) = ; / & \.

• Le service de transmission des données ne prend pas en charge l'utilisation d'une base de données OceanBase secondaire (standby) comme source.

Considérations importantes

• Si la source est OceanBase Database dans une version comprise entre V4.0.0 et V4.3.x (à l'exclusion de V4.2.5 BP1) et que vous avez sélectionné la synchronisation incrémentielle, spécifiez l'attribut STORED pour une colonne générée. Si vous ne spécifiez pas cet attribut, aucune information concernant la colonne générée n'est stockée dans les journaux incrémentiels, ce qui peut entraîner des erreurs de données lors de la synchronisation incrémentielle.

• Tenez compte des points suivants lorsqu'une ligne mise à jour contient une colonne de type objet volumineux (LOB) :

  • Si la colonne LOB est mise à jour, n'utilisez pas la valeur stockée dans la colonne LOB avant l'opération UPDATE ou DELETE.

    Les types de données suivants sont stockés dans les colonnes LOB : JSON, GIS, XML, type défini par l'utilisateur (UDT) et TEXT tels que LONGTEXT et MEDIUMTEXT.

  • Si la colonne LOB n'est pas mise à jour, la valeur stockée dans la colonne LOB avant et après l'opération UPDATE ou DELETE est NULL.

• Lors de la reprise d'un transfert de données pour une tâche, certaines données (datant de moins d'une minute) peuvent apparaître en double dans l'instance RocketMQ. Par conséquent, une déduplication est requise dans les systèmes en aval.

• Si les horloges ne sont pas synchronisées entre les nœuds ou entre le client et le serveur, la latence affichée lors de la synchronisation incrémentielle peut être inexacte.

  Par exemple, si l'horloge est en avance sur l'heure standard, la latence peut être négative. Si elle est en retard, la latence peut être positive.

• Si vous sélectionnez uniquement Incremental Synchronization lors de la création de la tâche de synchronisation des données, le service de transmission des données exige que les journaux incrémentiels locaux de la base de données source soient conservés pendant au moins 48 heures.

  Si vous sélectionnez Full Synchronization et Incremental Synchronization lors de la création de la tâche, le service de transmission des données exige que les journaux incrémentiels locaux de la base de données source soient conservés pendant au moins sept jours. À défaut, la tâche de synchronisation des données risque d'échouer ou les données des bases source et cible pourraient devenir incohérentes, car le service ne pourrait pas récupérer les journaux incrémentiels nécessaires.

Types d'instances source et cible pris en charge

Dans le tableau ci-dessous, OB_MySQL désigne un locataire MySQL d'OceanBase Database et OB_Oracle désigne un locataire Oracle d'OceanBase Database.

Procédure

1. Connectez-vous à la console de gestion OceanBase et achetez une tâche de synchronisation des données.

   Pour plus d'informations, consultez la section Acheter une tâche de synchronisation des données.

2. Dans la console Data Transmission Service, accédez à la page Data Synchronization, repérez la nouvelle tâche de synchronisation et cliquez sur Configure.

   

   Si vous souhaitez réutiliser la configuration d'une tâche existante, cliquez sur Reuse Configuration. Pour plus d'informations, consultez la section Réutiliser et effacer la configuration d'une tâche de synchronisation des données.

3. Sur la page Select Source and Target, configurez les paramètres.

   Paramètre	Description
   Nom de la tâche de synchronisation	Nous vous recommandons d'utiliser une combinaison de chiffres et de lettres. Le nom ne doit contenir aucun espace et sa longueur ne doit pas dépasser 64 caractères.
   Source	Si vous avez déjà créé une source de données OceanBase, sélectionnez-la dans la liste déroulante. Sinon, cliquez sur New Data Source dans la liste déroulante et créez-en une dans la boîte de dialogue qui s'affiche à droite. Pour plus d'informations sur les paramètres, consultez la section Créer une source de données OceanBase.
   Cible	Si vous avez déjà créé une source de données RocketMQ, sélectionnez-la dans la liste déroulante. Sinon, cliquez sur New Data Source dans la liste déroulante et créez-en une dans la boîte de dialogue qui s'affiche à droite. Pour plus d'informations, consultez la section Créer une source de données RocketMQ.
   Tag (facultatif)	Sélectionnez un tag cible dans la liste déroulante. Vous pouvez également cliquer sur Manage Tags pour créer, modifier ou supprimer des tags. Pour plus d'informations, consultez la section Utiliser des tags pour gérer les tâches de synchronisation des données.

4. Cliquez sur Next. Sur la page Select Synchronization Type, spécifiez les types de synchronisation pour la tâche actuelle.

   

   Les types de synchronisation pris en charge sont Full Synchronization et Incremental Synchronization. L'option Incremental Synchronization prend uniquement en charge la DML Synchronization. Les opérations DML prises en charge sont INSERT, DELETE et UPDATE. Sélectionnez les options selon vos besoins. Pour plus d'informations, consultez la section Configurer la synchronisation DDL/DML.

5. Cliquez sur Next. Sur la page Select Synchronization Objects, sélectionnez les objets à synchroniser dans la tâche actuelle.

   Utilisez l'option Specify Objects ou Match Rules pour spécifier les objets de synchronisation. Cette rubrique décrit l'utilisation de l'option Specify Objects. Pour savoir comment configurer les règles de correspondance, reportez-vous à la section « Modèles génériques pour la migration/synchronisation des données entre une base de données et une instance de file d'attente de messages » de la rubrique Configurer et modifier les règles de correspondance.

   Remarque

   Si vous avez sélectionné DDL Synchronization à l'étape Select Synchronization Type, nous vous recommandons de sélectionner les objets de synchronisation via l'option Match Rules. Cela garantit que tous les nouveaux objets répondant aux règles de correspondance seront synchronisés. Si vous utilisez l'option Specify Objects, les objets nouveaux ou renommés ne seront pas synchronisés.

   Lors de la synchronisation des données d'une base OceanBase vers une instance RocketMQ, vous pouvez sélectionner plusieurs tables à mapper vers plusieurs topics.

   1. Dans la section Select Synchronization Objects, sélectionnez Specify Objects.

   2. Dans le volet de gauche, sélectionnez les objets à synchroniser.

   3. Cliquez sur >.

   4. Dans la boîte de dialogue Map Object to Topic, ouvrez la liste déroulante Existing Topics et sélectionnez le topic cible.

      Vous pouvez également saisir le nom d'un topic existant et le sélectionner lorsqu'il apparaît.

      

   5. Cliquez sur OK.

      Le service de transmission des données vous permet d'importer des objets via un fichier texte. Il offre également la possibilité de modifier les topics associés aux objets, de définir des filtres de lignes et de supprimer un objet unique ou l'ensemble des objets. Les objets de la base de données cible sont listés selon la structure Topic > Database > Table.

      Remarque

      Lorsque vous sélectionnez Match Rules pour spécifier les objets de synchronisation, le renommage des objets est effectué selon la syntaxe des règles de correspondance définies. Dans la zone d'opérations, vous pouvez uniquement définir des conditions de filtrage et sélectionner les colonnes de sharding ainsi que les colonnes à synchroniser. Pour plus d'informations, consultez la section Configurer et modifier les règles de correspondance.

      

      Opération	Description
      Importer des objets	Dans la liste située à droite de la zone de sélection, cliquez sur Importer des objets en haut à droite. Dans la boîte de dialogue qui s'affiche, cliquez sur OK. Important L'importation d'objets écrase les sélections précédentes. Procédez avec prudence. Dans la boîte de dialogue Importer les objets de synchronisation, importez les objets à synchroniser. Vous pouvez importer un fichier CSV pour définir des conditions de filtrage de lignes, filtrer des colonnes et définir des colonnes de sharding. Pour plus d'informations, consultez la section Télécharger et importer les configurations des objets de synchronisation. Cliquez sur Valider. Une fois la validation réussie, cliquez sur OK.
      Modifier le topic	DTS vous permet de modifier le topic des objets de destination. Pour plus d'informations, consultez la section Modifier un topic.
      Paramètres	DTS prend en charge le filtrage des lignes à l'aide d'une clause WHERE et vous permet de sélectionner les colonnes de sharding ainsi que les colonnes à synchroniser. Dans la boîte de dialogue Settings, vous pouvez effectuer les opérations suivantes : Dans la zone Row Filter Condition, saisissez une clause SQL WHERE standard dans la zone de texte pour configurer le filtrage des lignes. Pour plus d'informations, consultez la section Filtrer les données à l'aide de conditions SQL. Dans la liste déroulante Sharding Column, sélectionnez la colonne de sharding cible. Vous pouvez sélectionner plusieurs champs comme colonnes de sharding. Ce paramètre est facultatif. Lors de la sélection des colonnes de sharding, si aucune exigence particulière ne s'applique, vous pouvez simplement sélectionner la clé primaire. Si la clé primaire entraîne un déséquilibre de charge, choisissez un champ unique et relativement équilibré comme colonne de sharding afin d'éviter d'éventuels problèmes de performance. Les principales fonctions des colonnes de sharding sont les suivantes : Équilibrage de charge : Si des écritures simultanées sont possibles sur la destination, la colonne de sharding sert à déterminer le thread spécifique utilisé pour l'envoi d'un message. Ordonnancement : Étant donné que les écritures simultanées peuvent provoquer des problèmes de désordre, DTS garantit que les messages ayant la même valeur de colonne de sharding sont reçus dans l'ordre. L'ordonnancement fait ici référence à l'ordre des modifications (l'ordre d'exécution des opérations DML sur une colonne). Dans la zone Select Columns, choisissez les colonnes à synchroniser. Pour plus d'informations, consultez la section Filtrage des colonnes.
      Supprimer/Supprimer tout	DTS vous permet de retirer un ou plusieurs objets temporairement sélectionnés pour la destination lors du mappage des données. Supprimer un seul objet de synchronisation Dans la liste située à droite de la zone de sélection, placez le curseur sur l'objet cible et cliquez sur le bouton Remove qui apparaît. Supprimer tous les objets de synchronisation Dans la liste située à droite de la zone de sélection, cliquez sur Remove All en haut à droite. Dans la boîte de dialogue, cliquez sur OK pour supprimer tous les objets de synchronisation.

6. Cliquez sur Next. Sur la page Synchronization Options, configurez les paramètres.

   • Synchronisation complète

     Les paramètres suivants s'affichent uniquement si vous sélectionnez Full Synchronization sur la page Choose Synchronization Types.

     

     Paramètre	Description
     Concurrence en lecture	Ce paramètre définit le nombre de threads simultanés utilisés pour lire les données depuis la source lors de la synchronisation complète. La valeur maximale est 512. Une concurrence élevée peut exercer une pression excessive sur la source et affecter votre activité.
     Concurrence en écriture	Ce paramètre définit le nombre de threads simultanés utilisés pour écrire les données vers la destination lors de la synchronisation complète. La valeur maximale est 512. Une concurrence élevée peut exercer une pression excessive sur la destination et affecter votre activité.
     Limitation du débit de synchronisation complète	Vous pouvez choisir d'activer ou non la limitation du débit de synchronisation complète selon vos besoins. Si vous l'activez, définissez les valeurs RPS (nombre maximal de lignes de données pouvant être synchronisées vers la destination par seconde lors de la synchronisation complète) et BPS (volume maximal de données pouvant être synchronisé vers la destination par seconde lors de la synchronisation complète). Remarque Les valeurs RPS et BPS définies ici servent uniquement à la limitation du débit. Les performances réelles de la synchronisation complète dépendent de facteurs tels que la source, la destination et les spécifications de l'instance.

   • Synchronisation incrémentielle

     Les paramètres suivants s'affichent uniquement si vous sélectionnez Incremental Synchronization sur la page Choose Synchronization Types.

     

     Paramètre	Description
     Concurrence en écriture	Ce paramètre définit le nombre de threads simultanés utilisés pour écrire les données vers la destination lors de la synchronisation incrémentielle. La valeur maximale est 512. Une concurrence élevée peut exercer une pression excessive sur la destination et affecter votre activité.
     Limitation du débit de synchronisation incrémentielle	Vous pouvez choisir d'activer ou non la limitation du débit de synchronisation incrémentielle selon vos besoins. Si vous l'activez, définissez les valeurs RPS (nombre maximal de lignes de données pouvant être synchronisées vers la destination par seconde lors de la synchronisation incrémentielle) et BPS (volume maximal de données pouvant être synchronisé vers la destination par seconde lors de la synchronisation incrémentielle). Remarque Les valeurs RPS et BPS définies ici servent uniquement à la limitation du débit. Les performances réelles de la synchronisation incrémentielle dépendent de facteurs tels que la source, la destination et les spécifications de l'instance.
     Décalage de départ de la synchronisation incrémentielle	Si vous sélectionnez Full Synchronization lors du choix du type de synchronisation, vous ne pouvez pas modifier ce paramètre. Si vous ne sélectionnez pas Full Synchronization mais choisissez Incremental Synchronization lors du choix du type de synchronisation, spécifiez un point dans le temps à partir duquel vous souhaitez synchroniser les données. La valeur par défaut est l'heure système actuelle. Pour plus d'informations, consultez la section Définir le décalage de départ pour la synchronisation incrémentielle.

   • Paramètres avancés

     

     Paramètre	Description
     Méthode de sérialisation	Format des messages pour la synchronisation des données vers l'instance RocketMQ cible. Les valeurs valides sont : Default, Canal, DataWorks (version 2.0 prise en charge), SharePlex, DefaultExtendColumnType, Debezium, DebeziumFlatten et DebeziumSmt. Pour plus d'informations, consultez la section Formats de données. Important Seuls les locataires MySQL d'OceanBase Database prennent en charge Debezium, DebeziumFlatten et DebeziumSmt.
     Règles de partitionnement	Lors de la synchronisation des données vers une instance RocketMQ, seule l'option Hash est prise en charge. L'option Hash indique que le service de transmission des données utilise un algorithme de hachage pour sélectionner une file d'attente de messages RocketMQ en fonction de la valeur de la clé primaire ou de la colonne de sharding.
     Identification du système métier (facultatif)	Identifie le système métier source des données. Ce paramètre s'affiche uniquement lorsque vous sélectionnez DataWorks pour l'option Serialization Method. L'identifiant du système métier doit comporter entre 1 et 20 caractères.

   • Cible

     

     Paramètre	Description
     Saisissez le nom du groupe de producteurs	Spécifie un groupe de producteurs capable d'écrire des données dans plusieurs topics.
     Autoriser le traçage des messages	Indique si le traçage des messages est autorisé. Si cette fonctionnalité est activée, vous pouvez suivre l'intégralité des informations de la tâche, y compris l'heure et le statut des nœuds, durant tout le cycle de vie d'un message, depuis son envoi vers le serveur RocketMQ par le producteur jusqu'à sa consommation par les consommateurs. La fonction de traçage des messages fournit un support de données pour le dépannage en environnement de production.

7. Cliquez sur Precheck.

   À l'étape Precheck, le service Data Transmission Service vérifie la connectivité entre la source et la destination. Si la prévalidation échoue :

   • Résolvez les problèmes identifiés, puis relancez la prévalidation.

   • Vous pouvez également cliquer sur Skip dans la colonne Actions pour un élément de prévalidation ayant échoué. Une boîte de dialogue s'affiche alors, détaillant l'impact spécifique de cette omission. Pour confirmer que vous souhaitez ignorer cet élément, cliquez sur OK dans la boîte de dialogue.

8. Une fois la prévalidation réussie, cliquez sur Start Task.

   Si vous ne souhaitez pas démarrer la tâche immédiatement, cliquez sur Save. Vous pourrez ensuite lancer la tâche manuellement depuis la page Sync Task List ou l'inclure dans une opération par lot. Pour plus d'informations sur les opérations par lot, consultez la section Effectuer des opérations par lot sur les tâches de synchronisation des données.

   DTS vous permet de modifier les objets de synchronisation pendant l'exécution d'une tâche de synchronisation des données. Pour plus d'informations, consultez la section Afficher et modifier les objets de synchronisation et leurs conditions de filtrage. Une fois la tâche de synchronisation des données lancée, elle s'exécute selon les types de synchronisation sélectionnés. Pour plus d'informations, consultez la section Afficher les détails de la synchronisation.

Si une tâche de synchronisation des données échoue, généralement en raison de problèmes de connexion réseau ou d'un démarrage lent du processus, cliquez sur Resume dans la liste des tâches ou sur la page du produit.

Références

• Créer une source de données OceanBase

• Créer une source de données RocketMQ

• Effectuer des opérations par lot sur les tâches de synchronisation des données

• Afficher les détails d'une tâche de synchronisation des données