AnalyticDB:Subscribe to binary logs by using Flink

Prérequis

• Votre cluster AnalyticDB for MySQL doit appartenir à l'une des éditions suivantes : Enterprise Edition, Basic Edition, Data Lakehouse Edition ou Data Warehouse Edition (en mode élastique).

• Le cluster AnalyticDB for MySQL doit respecter les exigences de version du noyau suivantes :

  • Moteur de table xuanwu_v1 : version 3.2.1.0 ou ultérieure.

  • Moteur de table xuanwu_v2 : version 3.2.5.0 ou ultérieure.

  • Pour vous abonner aux journaux binaires des vues matérialisées incrémentielles, la version du noyau doit être la 3.2.6.9, la 3.2.7.1 ou une version ultérieure.

  Remarque

  Pour afficher et mettre à jour la version mineure, accédez à la section Configuration Information de la page Cluster Information dans la console AnalyticDB for MySQL.

• L'espace de travail Flink doit utiliser Ververica Runtime (VVR) version 8.0.4 ou ultérieure.

• Le cluster AnalyticDB for MySQL et l'espace de travail Flink entièrement géré doivent se trouver dans le même VPC.

• Ajoutez le bloc CIDR de l'espace de travail Flink à la liste d'autorisation du service AnalyticDB for MySQL.

Limitations

• Flink ne peut traiter que les types de données de base et le type de données complexe JSON issus du journal binaire AnalyticDB for MySQL.

• Les opérations suivantes ne sont pas capturées lorsque vous utilisez Flink pour vous abonner aux journaux binaires AnalyticDB for MySQL. Les modifications de données correspondantes ne sont pas synchronisées vers les consommateurs en aval :

  • Opérations DDL : les modifications de schéma, telles que ALTER TABLE, ne sont pas capturées. Si le schéma de la table source change, vous devez mettre à jour manuellement la définition de la table dans la tâche Flink, puis redéployer celle-ci.

  • INSERT OVERWRITE : les données écrites via INSERT OVERWRITE ne génèrent pas d'enregistrements de journal binaire. Pour obtenir l'intégralité des données après une opération INSERT OVERWRITE, vous devez vous réabonner aux journaux binaires.

  • Éviction automatique des partitions : lorsque les partitions expirées d'une table partitionnée sont automatiquement supprimées, les données évictées ne génèrent pas d'enregistrements de journal binaire.

  • Actualisation complète des vues matérialisées : les opérations d'actualisation complète sur les vues matérialisées ne génèrent pas d'enregistrements de journal binaire.

    Remarque

    Les journaux binaires des vues matérialisées contiennent uniquement les modifications de données issues des actualisations incrémentielles. Pour obtenir l'état complet des données après une actualisation complète, vous devez vous réabonner aux journaux binaires.

Étape 3 : S'abonner au journal binaire

1. Connectez-vous à la console Realtime Compute for Apache Flink et créez une tâche SQL.

2. Créez une table source qui se connecte au service AnalyticDB for MySQL et lit les données du journal binaire d'une table spécifique (source_table).

   Remarque

   • La clé primaire définie dans le langage DDL Flink doit correspondre à la clé primaire de la table physique dans le cluster AnalyticDB for MySQL. Cela inclut à la fois les colonnes de clé et le nom de la clé primaire. Dans le cas contraire, les données risquent d'être incorrectes.

   • Les types de données Flink doivent être compatibles avec ceux du service AnalyticDB for MySQL. Pour plus de détails sur les mappages de types de données, consultez la section Mappage des types.

   CREATE TEMPORARY TABLE adb_source (
     `id` INT,
     `num` BIGINT,
     PRIMARY KEY (`id`) NOT ENFORCED
   ) WITH (
     'connector' = 'adb-mysql-cdc',
     'hostname' = 'amv-2zepb9n1l58ct01z50000****.ads.aliyuncs.com',
     'username' = 'testUser',
     'password' = 'Test12****',
     'database-name' = 'binlog',
     'table-name' = 'source_table'
   );

   Le tableau suivant décrit les paramètres WITH.

   Paramètre	Obligatoire	Valeur par défaut	Type	Description
   connector	Oui	Aucune	STRING	Le connecteur à utiliser. Il s'agit d'un connecteur personnalisé. Définissez ce paramètre sur adb-mysql-cdc.
   hostname	Oui	Aucune	STRING	Le point de terminaison AnalyticDB for MySQL du cluster AnalyticDB for MySQL.
   username	Oui	Aucune	STRING	Le compte de base de données du cluster AnalyticDB for MySQL.
   password	Oui	Aucune	STRING	Le mot de passe du compte de base de données AnalyticDB for MySQL.
   database-name	Oui	Aucune	STRING	Le nom de la base de données AnalyticDB for MySQL. Étant donné que AnalyticDB for MySQL implémente la journalisation binaire au niveau de la table, vous ne pouvez spécifier qu'une seule base de données.
   table-name	Oui	Aucune	STRING	Le nom de la table dans la base de données AnalyticDB for MySQL. Étant donné que AnalyticDB for MySQL implémente la journalisation binaire au niveau de la table, vous ne pouvez spécifier qu'une seule table.
   port	Non	3306	INTEGER	Le numéro de port.
   scan.incremental.snapshot.enabled	Non	true	BOOLEAN	Indique si le mécanisme de lecture par instantané incrémentiel est activé. Cette fonctionnalité est activée par défaut. Un instantané incrémentiel constitue un nouveau mécanisme de lecture des instantanés de table. Par rapport au mécanisme d'instantané précédent, les instantanés incrémentiels présentent les avantages suivants : La source prend en charge la lecture simultanée lors de la lecture d'instantanés. La source prend en charge les points de contrôle au niveau des fragments lors de la lecture d'instantanés. La source n'a pas besoin d'obtenir les autorisations de verrouillage de la base de données avant de lire un instantané.
   scan.incremental.snapshot.chunk.size	Non	8096	INTEGER	Le nombre de lignes par fragment pour un instantané de table.
   scan.snapshot.fetch.size	Non	1024	INTEGER	Le nombre maximal de lignes à récupérer à la fois lors de la lecture d'un instantané de table.
   scan.startup.mode	Non	initial	STRING	Le mode de démarrage pour la consommation des données. Valeurs valides : initial (par défaut) : effectue un instantané initial de la table, puis lit le dernier journal binaire. earliest-offset : ignore la phase d'instantané et commence la lecture à partir du journal binaire disponible le plus ancien. specific-offset : ignore la phase d'instantané et démarre à partir d'une position de journal binaire spécifiée. Spécifiez le nom du fichier de journal binaire et le décalage en définissant les paramètres scan.startup.specific-offset.file et scan.startup.specific-offset.pos. latest-offset : ignore la phase d'instantané et lit uniquement les modifications survenues après le démarrage du connecteur. timestamp : ignore la phase d'instantané et commence la lecture à partir d'un horodatage spécifié. Définissez l'horodatage en millisecondes (ms) à l'aide du paramètre scan.startup.timestamp-millis. Important Si vous utilisez le mode de démarrage earliest-offset, specific-offset ou timestamp, assurez-vous que le schéma de la table reste inchangé entre la position de départ spécifiée et le démarrage de la tâche. Dans le cas contraire, la tâche pourrait échouer en raison de modifications du schéma.
   scan.startup.specific-offset.file	Non	Aucune	STRING	En mode de démarrage specific-offset, le nom du fichier de journal binaire à la position de démarrage. Pour obtenir le dernier nom de fichier de journal binaire, exécutez l'instruction SHOW MASTER STATUS for table_name;.
   scan.startup.specific-offset.pos	Non	Aucune	LONG	En mode de démarrage specific-offset, la position dans le fichier de journal binaire à la position de démarrage. Pour obtenir la dernière position du journal binaire, exécutez l'instruction SHOW MASTER STATUS for table_name;.
   scan.startup.specific-offset.skip-events	Non	Aucune	LONG	Le nombre d'événements à ignorer après la position de démarrage spécifiée.
   scan.startup.specific-offset.skip-rows	Non	Aucune	LONG	Le nombre de lignes de données à ignorer après la position de démarrage spécifiée.
   scan.startup.timestamp-millis	Non	Aucune	LONG	L'horodatage en millisecondes de la position de démarrage lors de l'utilisation du mode de démarrage timestamp. Lorsque vous utilisez ce paramètre, vous devez définir scan.startup.mode sur timestamp. L'horodatage est exprimé en millisecondes (ms).
   server-time-zone	Non	Système par défaut	STRING	Le fuseau horaire de session sur le serveur de base de données, par exemple « Asia/Shanghai ». Il contrôle la façon dont AnalyticDB for MySQLAnalyticDB for MySQL est converti au type de données STRING. Si ce paramètre n'est pas défini, ZONELD.SYSTEMDEFAULT() est utilisé pour déterminer le fuseau horaire du serveur.
   debezium.min.row.count.to.stream.result	Non	1000	INTEGER	Si le nombre de lignes d'une table est supérieur à cette valeur, le connecteur diffuse les résultats en continu. Si vous définissez ce paramètre sur 0, toutes les vérifications de taille de table sont ignorées et tous les résultats sont diffusés en continu pendant la phase d'instantané.
   connect.timeout	Non	30s	DURATION	Le temps maximal d'attente d'une connexion à la base de données avant l'expiration de la tentative. L'unité par défaut est la seconde (s).
   connect.max-retries	Non	3	INTEGER	Le nombre maximal de tentatives après un échec de connexion à la base de données.

3. Créez une table physique dans la base de données de destination pour stocker les données traitées. Cette rubrique utilise le service AnalyticDB for MySQL comme destination. Pour connaître les connecteurs pris en charge par Flink, consultez la section Connecteurs pris en charge.

   CREATE TABLE target_table (
     `id` INT,
     `num` BIGINT,
     PRIMARY KEY (`id`)
   )

4. Créez une table de réception (sink) qui se connecte à la table créée à l'étape précédente. La table de réception écrit les données traitées dans la table spécifiée du service AnalyticDB for MySQL.

   CREATE TEMPORARY TABLE adb_sink (
     `id` INT,
     `num` BIGINT,
     PRIMARY KEY (`id`) NOT ENFORCED
   ) WITH (
     'connector' = 'adb3.0',
     'url' = 'jdbc:mysql://amv-2zepb9n1l58ct01z50000****.ads.aliyuncs.com:3306/flinktest',
     'userName' = 'testUser',
     'password' = 'Test12****',
     'tableName' = 'target_table'
   );

   Pour plus d'informations sur les paramètres WITH et les mappages de types pour la table de réception, consultez la section Connecteur AnalyticDB for MySQL V3.0.

5. Utilisez une instruction INSERT INTO pour envoyer les données de la table source vers la table de réception.

   INSERT INTO adb_sink
   SELECT * FROM adb_source;

6. Cliquez sur Save.

7. Cliquez sur Validate.

   La fonctionnalité de validation vérifie la sémantique SQL, la connectivité réseau et les métadonnées des tables utilisées par la tâche. Vous pouvez également cliquer sur SQL Advice dans la zone de résultats pour afficher les alertes de risque SQL et les suggestions d'optimisation.

8. (Facultatif) Cliquez sur Debug.

   Vous pouvez utiliser la fonctionnalité de débogage des tâches pour simuler l'exécution des tâches, vérifier les résultats de sortie, valider la logique métier des instructions SELECT ou INSERT, améliorer l'efficacité du développement et réduire les risques liés à la qualité des données.

9. Cliquez sur Deploy.

   Après avoir développé et validé la tâche, déployez-la dans l'environnement de production. Accédez ensuite à la page O&M et démarrez la tâche.

10. (Facultatif) Consultez les informations relatives au journal binaire.

    Remarque

    Les instructions suivantes renvoient 0 si vous avez activé la journalisation binaire sans toutefois vous y être encore abonné via DTS. Les informations de journal binaire n'apparaissent qu'après l'établissement réussi d'un abonnement.

    • Pour obtenir le nom de fichier et la position de la dernière entrée de journal binaire, exécutez l'instruction SQL suivante :

      SHOW MASTER STATUS FOR source_table;

    • Pour afficher tous les fichiers de journaux binaires non purgés et leur taille, exécutez l'instruction SQL suivante :

      SHOW BINARY LOGS FOR source_table;

Mappage des types

Le tableau suivant établit la correspondance entre les types de données AnalyticDB for MySQL et leurs équivalents Flink.