Connecteur MySQL YAML - Realtime Compute for Apache Flink - Centre de documentation Alibaba Cloud

Le connecteur MySQL peut être utilisé comme source de données dans les jobs d'ingestion de données au format YAML.

Prérequis

Avant d'utiliser une table source MySQL CDC, vous devez effectuer les opérations préalables décrites dans la rubrique Configurer MySQL.

ApsaraDB RDS for MySQL

Effectuez un test de connectivité réseau pour garantir la connectivité réseau avec Realtime Compute for Apache Flink.
Version MySQL : 5.6, 5,7, 8,0.x ou 8,4.
La journalisation binaire doit être activée. Elle est activée par défaut.
Le format du journal binaire doit être ROW. Il s'agit du format par défaut.
Le paramètre binlog_row_image doit être défini sur FULL. Il s'agit du paramètre par défaut.
La compression des transactions du journal binaire doit être désactivée. Cette fonctionnalité a été introduite dans MySQL 8.0.20 et est désactivée par défaut.
Un utilisateur MySQL a été créé avec les autorisations SELECT, SHOW DATABASES, REPLICATION SLAVE et REPLICATION CLIENT.
Créez une base de données et une table MySQL. Pour plus d'informations, consultez la rubrique Créer une base de données et un compte pour une instance ApsaraDB RDS for MySQL. Utilisez un compte privilégié pour créer la base de données MySQL afin d'éviter les échecs d'opération dus à des autorisations insuffisantes.
Configurez une liste d'autorisation d'adresses IP. Pour plus d'informations, consultez la rubrique Configurer une liste d'autorisation d'adresses IP pour une instance ApsaraDB RDS for MySQL.

PolarDB for MySQL

Effectuez un test de connectivité réseau pour garantir la connectivité réseau avec Realtime Compute for Apache Flink.
Version MySQL : 5.6, 5,7, 8,0.x ou 8,4.
La journalisation binaire doit être activée. Elle est désactivée par défaut.
Le format du journal binaire doit être ROW. Il s'agit du format par défaut.
Le paramètre binlog_row_image doit être défini sur FULL. Il s'agit du paramètre par défaut.
La compression des transactions du journal binaire doit être désactivée. Cette fonctionnalité a été introduite dans MySQL 8.0.20 et est désactivée par défaut.
Vous avez créé un utilisateur MySQL disposant des autorisations SELECT, SHOW DATABASES, REPLICATION SLAVE et REPLICATION CLIENT.
Créez une base de données et une table MySQL. Pour plus d'informations, consultez la rubrique Créer une base de données et un compte pour un cluster PolarDB for MySQL. Utilisez un compte privilégié pour créer la base de données MySQL afin d'éviter les échecs d'opération dus à des autorisations insuffisantes.
Configurez une liste d'autorisation d'adresses IP. Pour plus d'informations, consultez la rubrique Configurer une liste d'autorisation d'adresses IP pour un cluster PolarDB for MySQL.

Self-managed MySQL

Effectuez un test de connectivité réseau pour garantir la connectivité réseau avec Realtime Compute for Apache Flink.
Version MySQL : 5.6, 5,7, 8,0.x ou 8,4.
La journalisation binaire doit être activée. Elle est désactivée par défaut.
Le format du journal binaire doit être ROW. Le format par défaut est STATEMENT.
Le paramètre binlog_row_image doit être défini sur FULL. Il s'agit du paramètre par défaut.
La compression des transactions du journal binaire doit être désactivée. Cette fonctionnalité a été introduite dans MySQL 8.0.20 et est désactivée par défaut.
Créez un utilisateur MySQL et accordez-lui les autorisations SELECT, SHOW DATABASES, REPLICATION SLAVE et REPLICATION CLIENT.
Créez une base de données et une table MySQL. Pour plus d'informations, consultez la documentation Créer une base de données et un compte pour une instance MySQL auto-gérée. Utilisez un compte privilégié pour créer la base de données MySQL afin d'éviter les échecs d'opération dus à des autorisations insuffisantes.
Configurez une liste d'autorisation d'adresses IP. Pour plus d'informations, consultez la rubrique Configurer une liste d'autorisation d'adresses IP pour une instance MySQL auto-gérée.

Limites

Limites générales

Le connecteur MySQL CDC ne prend pas en charge la fonctionnalité de compression des transactions du journal binaire. Par conséquent, lorsque vous utilisez le connecteur MySQL CDC pour consommer des données incrémentielles, assurez-vous que la compression des transactions du journal binaire est désactivée. Sinon, le connecteur risque de ne pas pouvoir récupérer les données incrémentielles.

Limites d'ApsaraDB RDS for MySQL

Pour ApsaraDB RDS for MySQL, ne lisez pas les données depuis une base de données secondaire ou un réplica en lecture seule. En effet, la période de rétention par défaut des journaux binaires pour les bases de données secondaires et les réplicas en lecture seule est courte. Si les journaux binaires expirent et sont supprimés, le job peut échouer à consommer les données du journal binaire et générer une erreur.
ApsaraDB RDS for MySQL active par défaut la synchronisation parallèle primaire/secondaire, mais ne garantit pas un ordre de transaction cohérent entre les instances primaires et secondaires. Cela peut entraîner une perte de données lors d'un basculement primaire/secondaire et d'une récupération à partir d'un point de contrôle. Pour éviter ce problème, vous pouvez activer manuellement l'option slave_preserve_commit_order pour ApsaraDB RDS for MySQL.

Limites de PolarDB for MySQL

Les tables sources MySQL CDC ne prennent pas en charge la lecture des données depuis les clusters d'architecture Multi-master Cluster de PolarDB for MySQL V1.0.19 et versions antérieures. Pour plus d'informations, consultez la rubrique Qu'est-ce qu'un cluster Multi-master ?. Les journaux binaires générés par ces clusters peuvent contenir des ID de table en double. Cela peut provoquer des erreurs de mappage de schéma dans la table source CDC, entraînant des erreurs lors de l'analyse des données du journal binaire.

Limites de MySQL open source

Par défaut, MySQL maintient l'ordre des transactions lors de la réplication des journaux binaires entre le serveur primaire et le serveur secondaire. Si un réplica MySQL a la réplication parallèle activée (slave_parallel_workers > 1) mais n'a pas activé slave_preserve_commit_order=ON, l'ordre de validation des transactions peut être incohérent avec celui de la base de données primaire. Lorsque Flink CDC récupère à partir d'un point de contrôle, il peut manquer des données en raison du séquençage désordonné. Vous pouvez définir slave_preserve_commit_order = ON sur le réplica MySQL. Vous pouvez également définir slave_parallel_workers = 1, mais cela réduira les performances de réplication.

Remarques d'utilisation

Définir un ID de serveur pour éviter les conflits de consommation du journal binaire.
Pendant la phase de lecture complète des données, vous ne pouvez pas enregistrer de point de contrôle, ajouter une nouvelle table à la table source ou en supprimer une, puis redémarrer le job à partir du point de contrôle. Cela empêcherait le job de lire les données.

Ingestion de données

Vous pouvez utiliser le connecteur MySQL comme source de données dans un job d'ingestion de données au format YAML.

Syntaxe

source:
   type: mysql
   name: MySQL Source
   hostname: localhost
   port: 3306
   username: <username>
   password: <password>
   tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
   server-id: 5401-5404

sink:
  type: xxx

Éléments de configuration

Paramètre	Description	Obligatoire	Type de données	Valeur par défaut	Notes
type	Le type de source de données.	Oui	STRING	Aucune	La valeur doit être mysql.
name	Le nom de la source de données.	Non	STRING	Aucune	Aucune.
hostname	L'adresse IP ou le nom d'hôte de la base de données MySQL.	Oui	STRING	Aucune	Nous vous recommandons de spécifier une adresse VPC. Remarque Si la base de données MySQL et Realtime Compute for Apache Flink ne se trouvent pas dans le même VPC, vous devez établir une connexion réseau inter-VPC ou utiliser un endpoint public pour accéder à la base de données. Pour plus d'informations, consultez Gérer et exploiter les espaces de travail et Comment un cluster Flink entièrement géré peut-il accéder à Internet ?.
username	Le nom d'utilisateur du service de base de données MySQL.	Oui	STRING	Aucune	Aucune.
password	Le mot de passe du service de base de données MySQL.	Oui	STRING	Aucune	Aucune.
tables	Les tables de données MySQL à synchroniser.	Oui	STRING	Aucune	Ce paramètre prend en charge les expressions régulières pour lire les données de plusieurs tables. Vous pouvez utiliser des virgules pour séparer plusieurs expressions régulières. Remarque N'utilisez pas les caractères de correspondance de début de chaîne `^` et de fin de chaîne `$` dans l'expression régulière. Dans VVR 11.2, le point est utilisé pour diviser l'expression régulière afin d'obtenir la partie base de données. Les caractères de correspondance de début et de fin rendront l'expression régulière de la base de données résultante inutilisable. Par exemple, vous devez remplacer `^db.user_[0-9]+$` par `db.user_[0-9]+`. Le point est utilisé pour séparer le nom de la base de données et le nom de la table. Pour utiliser un point afin de correspondre à n'importe quel caractère, vous devez l'échapper avec une barre oblique inverse. Par exemple : db0.\., db1.user_table_[0-9]+, db[1-2].[app\|web]order_\..
tables.exclude	Les tables à exclure de la synchronisation.	Non	STRING	Aucune	Ce paramètre prend en charge les expressions régulières pour exclure plusieurs tables. Vous pouvez utiliser des virgules pour séparer plusieurs expressions régulières. Remarque Le point est utilisé pour séparer le nom de la base de données et le nom de la table. Pour utiliser un point afin de correspondre à n'importe quel caractère, vous devez l'échapper avec une barre oblique inverse. Par exemple : db0.\., db1.user_table_[0-9]+, db[1-2].[app\|web]order_\..
port	Le numéro de port du service de base de données MySQL.	Non	INTEGER	3306	Aucune.
schema-change.enabled	Indique s'il faut envoyer des événements de modification de schéma.	Non	BOOLEAN	true	Aucune.
server-id	Un ID numérique ou une plage pour le client de base de données utilisé pour la synchronisation.	Non	STRING	Une valeur aléatoire entre 5400 et 6400 est générée.	Cet ID doit être globalement unique au sein du cluster MySQL. Définissez un ID différent pour chaque tâche qui se connecte à la même base de données. Ce paramètre prend également en charge un format de plage d'ID, tel que 5400-5408. Remarque Lorsque la lecture incrémentielle est activée, la lecture simultanée est prise en charge. Dans ce cas, nous vous recommandons de définir une plage d'ID afin que chaque lecteur simultané utilise un ID différent.
jdbc.properties.*	Paramètres de connexion personnalisés dans l'URL JDBC.	Non	STRING	Aucune	Vous pouvez transmettre des paramètres de connexion personnalisés. Par exemple, pour ne pas utiliser le protocole SSL, vous pouvez configurer 'jdbc.properties.useSSL' = 'false'. Pour plus d'informations sur les paramètres de connexion pris en charge, consultez Propriétés de configuration MySQL.
debezium.*	Paramètres personnalisés pour Debezium afin de lire les journaux binaires.	Non	STRING	Aucune	Vous pouvez transmettre des paramètres Debezium personnalisés. Par exemple, utilisez 'debezium.event.deserialization.failure.handling.mode'='ignore' pour spécifier la logique de gestion des erreurs d'analyse. Avertissement Ne modifiez pas arbitrairement les paramètres Debezium. Cela pourrait entraîner une lecture incorrecte des données par le connecteur. Par exemple, le paramètre debezium.binlog.buffer.size ne doit pas être configuré.
scan.incremental.snapshot.chunk.size	La taille de chaque fragment en nombre de lignes.	Non	INTEGER	8096	Les tables MySQL sont divisées en plusieurs fragments pour la lecture. Les données d'un fragment sont mises en cache en mémoire avant d'être entièrement lues. Moins un fragment contient de lignes, plus le nombre total de fragments dans la table est élevé. Bien que cela réduise la granularité de la récupération après incident, cela peut entraîner des erreurs OOM (Out Of Memory) et diminuer le débit global. Par conséquent, vous devez trouver un compromis et définir une taille de fragment raisonnable.
scan.snapshot.fetch.size	Le nombre maximal d'enregistrements à extraire à la fois lors de la lecture des données complètes d'une table.	Non	INTEGER	1024	Aucune.
scan.startup.mode	Le mode de démarrage pour la consommation des données.	Non	STRING	initial	Valeurs valides : initial (par défaut) : Au premier démarrage ou lors d'un démarrage sans état, le connecteur analyse toutes les données historiques, puis lit les dernières données du journal binaire. latest-offset : Au premier démarrage ou lors d'un démarrage sans état, le connecteur n'analyse pas les données historiques. Il commence la lecture à partir de la fin du journal binaire, ce qui signifie qu'il ne lit que les dernières modifications apportées après le démarrage du connecteur. earliest-offset : Le connecteur n'analyse pas les données historiques. Il commence la lecture à partir du journal binaire disponible le plus ancien. specific-offset : Le connecteur n'analyse pas les données historiques. Il démarre à partir d'un décalage spécifique du journal binaire. Vous pouvez spécifier le décalage en configurant à la fois scan.startup.specific-offset.file et scan.startup.specific-offset.pos, ou en configurant uniquement scan.startup.specific-offset.gtid-set pour démarrer à partir d'un ensemble GTID spécifique. timestamp : Le connecteur n'analyse pas les données historiques. Il commence à lire le journal binaire à partir d'un horodatage spécifié. L'horodatage est spécifié par scan.startup.timestamp-millis en millisecondes. Important Pour les modes de démarrage earliest-offset, specific-offset et timestamp, si le schéma de la table au moment du démarrage diffère du schéma à l'heure du décalage de démarrage spécifié, la tâche signalera une erreur en raison de l'incompatibilité des schémas. Autrement dit, lorsque vous utilisez ces trois modes de démarrage, vous devez vous assurer que le schéma de la table correspondante ne change pas entre la position de consommation du journal binaire spécifiée et l'heure de démarrage de la tâche.
scan.startup.specific-offset.file	Le nom du fichier de journal binaire pour le décalage de démarrage lors de l'utilisation du mode de démarrage specific-offset.	Non	STRING	Aucune	Lorsque vous utilisez ce paramètre, vous devez définir scan.startup.mode sur specific-offset. Exemple de format de nom de fichier : `mysql-bin.000003`.
scan.startup.specific-offset.pos	Le décalage dans le fichier de journal binaire spécifié pour le décalage de démarrage lors de l'utilisation du mode de démarrage specific-offset.	Non	INTEGER	Aucune	Lorsque vous utilisez ce paramètre, vous devez définir scan.startup.mode sur specific-offset.
scan.startup.specific-offset.gtid-set	L'ensemble GTID pour le décalage de démarrage lors de l'utilisation du mode de démarrage specific-offset.	Non	STRING	Aucune	Lorsque vous utilisez ce paramètre, vous devez définir scan.startup.mode sur specific-offset. Exemple de format d'ensemble GTID : `24DA167-0C0C-11E8-8442-00059A3C7B00:1-19`.
scan.startup.timestamp-millis	L'horodatage en millisecondes pour le décalage de démarrage lors de l'utilisation du mode de démarrage timestamp.	Non	LONG	Aucune	Lorsque vous utilisez ce paramètre, vous devez définir scan.startup.mode sur timestamp. L'unité de l'horodatage est la milliseconde. Important Lorsque vous spécifiez une heure, MySQL CDC tente de lire l'événement initial de chaque fichier de journal binaire pour déterminer son horodatage. Il localise ensuite le fichier de journal binaire correspondant à l'heure spécifiée. Assurez-vous que le fichier de journal binaire correspondant à l'horodatage spécifié n'a pas été effacé de la base de données et qu'il peut être lu.
server-time-zone	Le fuseau horaire de session utilisé par la base de données.	Non	STRING	Si vous ne spécifiez pas ce paramètre, le système utilise le fuseau horaire de l'environnement d'exécution de la tâche Flink comme fuseau horaire du serveur de base de données. Il s'agit du fuseau horaire de la zone que vous avez sélectionnée.	Exemple : Asia/Shanghai. Ce paramètre contrôle la conversion du type TIMESTAMP dans MySQL vers le type STRING. Pour plus d'informations, consultez Valeurs temporelles Debezium.
scan.startup.specific-offset.skip-events	Le nombre d'événements de journal binaire à ignorer lors de la lecture à partir d'un décalage spécifié.	Non	INTEGER	Aucune	Lorsque vous utilisez ce paramètre, vous devez définir scan.startup.mode sur specific-offset.
scan.startup.specific-offset.skip-rows	Le nombre de modifications de lignes à ignorer lors de la lecture à partir d'un décalage spécifié. Un seul événement de journal binaire peut correspondre à plusieurs modifications de lignes.	Non	INTEGER	Aucune	Lorsque vous utilisez ce paramètre, vous devez définir scan.startup.mode sur specific-offset.
connect.timeout	Le temps maximal d'attente avant expiration de la connexion au serveur de base de données MySQL avant nouvelle tentative.	Non	DURATION	30 s	Aucune.
connect.max-retries	Le nombre maximal de tentatives après un échec de connexion au service de base de données MySQL.	Non	INTEGER	3	Aucune.
connection.pool.size	La taille du pool de connexions à la base de données.	Non	INTEGER	20	Le pool de connexions à la base de données est utilisé pour réutiliser les connexions, ce qui peut réduire le nombre de connexions à la base de données.
heartbeat.interval	L'intervalle auquel la source fait avancer le décalage du journal binaire à l'aide d'événements de pulsation.	Non	DURATION	30s	Les événements de pulsation sont utilisés pour faire avancer le décalage du journal binaire dans la source. Cela est très utile pour les tables MySQL qui sont mises à jour rarement. Pour de telles tables, le décalage du journal binaire ne peut pas avancer automatiquement. Les événements de pulsation peuvent faire avancer le décalage du journal binaire, ce qui évite les problèmes causés par un décalage de journal binaire expiré. Un décalage de journal binaire expiré peut entraîner l'échec de la tâche et la rendre irrécupérable, nécessitant un redémarrage sans état.
rds.region-id	L'ID de région de l'instance ApsaraDB RDS for MySQL d'Alibaba Cloud.	Obligatoire lors de l'utilisation de la fonctionnalité de lecture des journaux archivés depuis OSS.	STRING	Aucune	Pour plus d'informations sur les ID de région, consultez Régions et zones. Important Étant donné que la chaîne GTID pour MySQL CDC est générée de manière aléatoire et n'est pas monotone croissante comme les décalages de fichiers de journaux binaires, la localisation d'un GTID dans un fichier nécessite le téléchargement et l'analyse de tous les journaux archivés depuis OSS. Ce processus consomme beaucoup de ressources et prend du temps, rendant impossibles les fonctionnalités qui reposent sur les décalages GTID. Par conséquent, la fonctionnalité de journaux archivés OSS prend uniquement en charge le démarrage à partir d'un horodatage spécifié ou d'un décalage de fichier de journal binaire spécifié. Elle ne prend pas en charge le démarrage à partir d'un GTID spécifié, ni les scénarios impliquant des basculements primaire/secondaire dans les journaux archivés, car les basculements primaire/secondaire MySQL reposent sur les GTID. Évaluez attentivement cette fonctionnalité avant utilisation.
rds.access-key-id	L'AccessKey ID du compte ApsaraDB RDS for MySQL d'Alibaba Cloud.	Obligatoire lors de l'utilisation de la fonctionnalité de lecture des journaux archivés depuis OSS.	STRING	Aucune	Pour plus d'informations, consultez Comment afficher l'AccessKey ID et l'AccessKey secret ? Important Pour éviter la divulgation de vos informations AccessKey, utilisez la fonctionnalité de gestion des secrets pour spécifier l'AccessKey ID. Pour plus d'informations, consultez Gérer les variables.
rds.access-key-secret	L'AccessKey secret du compte ApsaraDB RDS for MySQL d'Alibaba Cloud.	Obligatoire lors de l'utilisation de la fonctionnalité de lecture des journaux archivés depuis OSS.	STRING	Aucune	Pour plus d'informations, consultez Comment afficher l'AccessKey ID et l'AccessKey secret ? Important Pour éviter la divulgation de vos informations AccessKey, utilisez la fonctionnalité de gestion des secrets pour spécifier l'AccessKey secret. Pour plus d'informations, consultez Gérer les variables.
rds.db-instance-id	L'ID de l'instance ApsaraDB RDS for MySQL d'Alibaba Cloud.	Obligatoire lors de l'utilisation de la fonctionnalité de lecture des journaux archivés depuis OSS.	STRING	Aucune	Aucune.
rds.main-db-id	Le numéro de la base de données principale de l'instance ApsaraDB RDS for MySQL d'Alibaba Cloud.	Non	STRING	Aucune	Pour plus d'informations sur l'obtention du numéro de la base de données principale, consultez Sauvegarde des journaux ApsaraDB RDS for MySQL. Remarque Si ce paramètre n'est pas spécifié, VVR 11.7 et versions ultérieures interrogent automatiquement le numéro de la base de données principale en fonction des informations de connexion ApsaraDB RDS for MySQL.
rds.download.timeout	Le délai d'expiration pour le téléchargement d'un seul journal archivé depuis OSS.	Non	DURATION	60s	Aucune.
rds.endpoint	L'endpoint de service pour obtenir les informations sur les journaux binaires OSS.	Non	STRING	Aucune	Pour plus d'informations sur les valeurs valides, consultez Endpoints.
rds.binlog-directory-prefix	Le préfixe du répertoire pour stocker les fichiers de journaux binaires.	Non	STRING	rds-binlog-	Aucune.
rds.use-intranet-link	Indique s'il faut utiliser le réseau interne pour télécharger les fichiers de journaux binaires.	Non	BOOLEAN	true	Aucune.
rds.binlog-directories-parent-path	Le chemin absolu du répertoire parent pour stocker les fichiers de journaux binaires.	Non	STRING	Aucune	Aucune.
chunk-meta.group.size	La taille des métadonnées de fragment.	Non	INTEGER	1000	Si les métadonnées dépassent cette valeur, elles sont divisées en plusieurs parties pour la transmission.
chunk-key.even-distribution.factor.lower-bound	La limite inférieure du facteur de distribution des fragments pour un partitionnement uniforme.	Non	DOUBLE	0.05	Si le facteur de distribution est inférieur à cette valeur, un partitionnement non uniforme est utilisé. Facteur de distribution des fragments = (MAX(clé-de-fragment) - MIN(clé-de-fragment) + 1) / Nombre total de lignes de données.
chunk-key.even-distribution.factor.upper-bound	La limite supérieure du facteur de distribution des fragments pour un partitionnement uniforme.	Non	DOUBLE	1000.0	Si le facteur de distribution est supérieur à cette valeur, un partitionnement non uniforme est utilisé. Facteur de distribution des fragments = (MAX(clé-de-fragment) - MIN(clé-de-fragment) + 1) / Nombre total de lignes de données.
scan.incremental.close-idle-reader.enabled	Indique s'il faut fermer les lecteurs inactifs une fois l'instantané terminé.	Non	BOOLEAN	false	Pour que cette configuration prenne effet, vous devez définir `execution.checkpointing.checkpoints-after-tasks-finish.enabled` sur true.

scan.only.deserialize.captured.tables.changelog.enabled	En phase incrémentale, indique s'il faut désérialiser uniquement les événements de modification des tables spécifiées.	Non	BOOLEAN	La valeur par défaut est false dans les versions VVR 8.x. La valeur par défaut est true à partir de la version VVR 11.1.	Valeurs valides : true : Désérialise uniquement les données de modification des tables cibles pour accélérer la lecture du journal binaire. false (par défaut) : Désérialise les données de modification de toutes les tables.
scan.parallel-deserialize-changelog.enabled	En phase incrémentale, indique s'il faut utiliser plusieurs threads pour analyser les événements de modification.	Non	BOOLEAN	false	Valeurs valides : true : Utilise plusieurs threads lors de la phase de désérialisation des événements de modification tout en maintenant l'ordre des événements du journal binaire pour accélérer la lecture. false (par défaut) : Utilise un seul thread lors de la phase de désérialisation des événements. Remarque Pris en charge uniquement à partir de la version VVR 8.0.11.
scan.parallel-deserialize-changelog.handler.size	Le nombre de gestionnaires d'événements lors de l'utilisation de plusieurs threads pour analyser les événements de modification.	Non	INTEGER	2	Remarque Pris en charge uniquement à partir de la version VVR 8.0.11.
metadata-column.include-list	Les colonnes de métadonnées à transmettre au flux en aval.	Non	STRING	Aucune	Les métadonnées disponibles incluent `op_ts`, `es_ts`, `query_log`, `file` et `pos`. Vous pouvez séparer plusieurs colonnes de métadonnées par des virgules. Remarque Le connecteur YAML MySQL CDC ne nécessite pas et ne prend pas en charge l'ajout des colonnes de métadonnées pour le nom de la base de données, le nom de la table et `op_type`. Vous pouvez utiliser directement `__data_event_type__` dans une expression Transform pour obtenir le type de données de modification, ou utiliser `__schema_name__` et `__table_name__` pour obtenir le nom de la base de données et le nom de la table. Important La colonne de métadonnées `file` représente le fichier de journal binaire où se trouvent les données. Elle est vide (« "" ») pendant la phase complète et contient le nom du fichier de journal binaire pendant la phase incrémentale. La colonne de métadonnées `pos` représente le décalage des données dans le fichier de journal binaire. Elle est égale à « 0 » pendant la phase complète et contient le décalage des données dans le fichier de journal binaire pendant la phase incrémentale. Ces deux colonnes de métadonnées sont prises en charge à partir de la version VVR 11.5. La colonne de métadonnées `es_ts` représente l'heure de début de la transaction correspondante pour le journal des modifications sur MySQL. Elle est prise en charge uniquement pour MySQL 8.0.x. N'ajoutez pas cette colonne de métadonnées si vous utilisez des versions antérieures de MySQL. L'horodatage `op_ts` est précis à la seconde près, tandis que l'horodatage `es_ts` est précis à la milliseconde près.
scan.newly-added-table.enabled	Lors d'un redémarrage à partir d'un point de contrôle, indique s'il faut synchroniser les tables nouvellement ajoutées qui n'ont pas été mises en correspondance lors du démarrage précédent, ou supprimer de l'état les tables qui ne correspondent plus.	Non	BOOLEAN	false	Cela prend effet lors d'un redémarrage à partir d'un point de contrôle ou d'un point de sauvegarde. Important Pendant la phase de lecture complète des données, vous ne pouvez pas enregistrer de point de sauvegarde, ajouter une nouvelle table à la table source ou en supprimer une, puis redémarrer le job à partir du point de sauvegarde. Cela entraînerait l'échec de la lecture des données par le job.
scan.binlog.newly-added-table.enabled	En phase incrémentale, indique s'il faut envoyer les données des tables nouvellement ajoutées qui correspondent aux critères.	Non	BOOLEAN	false	Ne peut pas être activé en même temps que `scan.newly-added-table.enabled`.
scan.incremental.snapshot.chunk.key-column	Spécifie une colonne pour certaines tables à utiliser comme colonne de division pour le sharding pendant la phase d'instantané.	Non	STRING	Aucune	Utilisez deux-points `:` pour connecter le nom de la table et le nom de la colonne afin de définir une règle. Le nom de la table peut être une expression régulière. Vous pouvez définir plusieurs règles en les séparant par un point-virgule `;`. Par exemple : `db1.user_table_[0-9]+:col1;db[1-2].[app\|web]_order_\\.*:col2`. Requis pour les tables sans clé primaire. La colonne sélectionnée doit être de type non nul (NOT NULL). Facultatif pour les tables avec une clé primaire. Une seule colonne peut être sélectionnée parmi la clé primaire.
scan.parse.online.schema.changes.enabled	En phase incrémentale, indique s'il faut tenter d'analyser les événements DDL de modification sans verrou RDS.	Non	BOOLEAN	false	Valeurs valides : true : Analyse les événements DDL de modification sans verrou RDS. false (par défaut) : N'analyse pas les événements DDL de modification sans verrou RDS. Il s'agit d'une fonctionnalité expérimentale. Avant d'effectuer une modification en ligne sans verrou, prenez un instantané du job Flink pour la récupération. Remarque Pris en charge uniquement à partir de la version VVR 11.0.
scan.incremental.snapshot.backfill.skip	Indique s'il faut ignorer le remplissage rétroactif (backfill) pendant la phase de lecture d'instantané.	Non	BOOLEAN	false	Valeurs valides : true : Ignore le remplissage rétroactif pendant la phase de lecture d'instantané. false (par défaut) : N'ignore pas le remplissage rétroactif pendant la phase de lecture d'instantané. Le remplissage rétroactif s'applique uniquement lors de la requête d'instantané d'un seul fragment (chunk) et ne couvre pas l'intégralité de la phase de lecture complète. Lorsque le remplissage rétroactif est ignoré, la requête d'instantané de chaque fragment lit les dernières données de la table à cet instant précis ; les mises à jour qui surviennent sur un fragment après sa lecture ne sont pas fusionnées pendant la phase de lecture complète et sont lues depuis le journal binaire (Binlog) après l'entrée en phase incrémentale. Par exemple, une mise à jour du fragment 5 qui se produit pendant la prise d'instantané du fragment 5 est reflétée directement dans l'instantané du fragment 5 ; si le fragment 5 est mis à jour après que le lecteur est passé au fragment 80, la mise à jour est appliquée ultérieurement depuis le journal binaire pendant la phase incrémentale. Important Lorsqu'elle est activée, les modifications qui surviennent pendant ou après l'analyse d'un fragment sont toujours transmises depuis le journal binaire en phase incrémentale et peuvent entraîner des doublons. Seule la sémantique « au moins une fois » (at-least-once) est garantie. Activez cette option uniquement si le puits (sink) en aval prend en charge les écritures idempotentes par clé primaire. Remarque Pris en charge uniquement à partir de la version VVR 11.1.
treat-tinyint1-as-boolean.enabled	Indique s'il faut traiter le type TINYINT(1) comme un type booléen.	Non	BOOLEAN	true	Valeurs valides : true (par défaut) : Traite le type TINYINT(1) comme un type booléen. false : Ne traite pas le type TINYINT(1) comme un type booléen.
treat-timestamp-as-datetime-enabled	Indique s'il faut traiter le type TIMESTAMP comme un type DATETIME.	Non	BOOLEAN	false	Valeurs valides : true : Traite le type MySQL TIMESTAMP comme un type DATETIME et le mappe au type CDC TIMESTAMP. false (par défaut) : Mappe le type MySQL TIMESTAMP au type CDC TIMESTAMP_LTZ. Le type MySQL TIMESTAMP stocke l'heure UTC et est affecté par le fuseau horaire. Le type MySQL DATETIME stocke l'heure littérale et n'est pas affecté par le fuseau horaire. Lorsqu'il est activé, ce paramètre convertit les données de type MySQL TIMESTAMP en type DATETIME en fonction du fuseau horaire du serveur (server-time-zone).
include-comments.enabled	Indique s'il faut synchroniser les commentaires des tables et des colonnes.	Non	BOOELEAN	false	Valeurs valides : true : Synchronise les commentaires des tables et des colonnes. false (par défaut) : Ne synchronise pas les commentaires des tables et des colonnes. L'activation de cette option augmente l'utilisation de la mémoire du job.
scan.incremental.snapshot.unbounded-chunk-first.enabled	Indique s'il faut distribuer en priorité les fragments non bornés (unbounded chunks) pendant la phase de lecture d'instantané.	Non	BOOELEAN	false	Valeurs valides : true : Distribue en priorité les fragments non bornés pendant la phase de lecture d'instantané. false (par défaut) : Ne distribue pas en priorité les fragments non bornés pendant la phase de lecture d'instantané. Il s'agit d'une fonctionnalité expérimentale. Son activation peut réduire le risque d'erreurs de dépassement de mémoire (OOM) sur le TaskManager lors de la synchronisation du dernier fragment pendant la phase d'instantané. Ajoutez ce paramètre avant le premier démarrage du job. Remarque Pris en charge uniquement à partir de la version VVR 11.1.
binlog.session.network.timeout	Le délai d'expiration réseau pour la connexion au journal binaire.	Non	DURATION	10m	Si la valeur est définie sur 0s, le délai d'expiration par défaut du serveur MySQL est utilisé. Remarque Pris en charge uniquement à partir de la version VVR 11.5.
scan.rate-limit.records-per-second	Limite le nombre maximal d'enregistrements envoyés par la source par seconde.	Non	LONG	Aucune	Cela s'applique aux scénarios où la lecture des données doit être limitée. Cette limite est effective dans les phases complète et incrémentale. La métrique `numRecordsOutPerSecond` de la source reflète le nombre d'enregistrements produits par l'ensemble du flux de données par seconde. Vous pouvez ajuster ce paramètre en fonction de cette métrique. Pendant la phase de lecture complète des données, vous devez généralement réduire le nombre de lignes lues dans chaque lot. Vous pouvez réduire la valeur du paramètre `scan.incremental.snapshot.chunk.size`. Remarque Pris en charge uniquement à partir de la version VVR 11.5.
include-binlog-meta.enable	Indique s'il faut inclure les informations originales du journal binaire MySQL, telles que le GTID et le décalage du journal binaire, dans le message.	Non	Boolean	false	Cela s'applique aux scénarios de synchronisation du journal binaire original, tels que le remplacement d'un lien de synchronisation Canal existant. Remarque Pris en charge uniquement à partir de la version VVR 11.6.
scan.binlog.tolerate.gtid-holes	L'activation de ce paramètre ignore les lacunes dans la séquence GTID, permettant au job de contourner les événements discontinus et de continuer à s'exécuter.	Non	Boolean	false	Avant d'activer ce paramètre, vous devez vous assurer que le décalage de départ du job n'a pas expiré. Si le job démarre à partir d'un décalage GTID effacé ou expiré, le moteur ignorera finalement les journaux manquants, ce qui entraînera une perte de données. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.6.
scan.emit.create-table-events.in-batch.enabled	Indique s'il faut envoyer les schémas de table par lots pendant la phase d'initialisation du job.	Non	Boolean	false	Il s'agit d'une fonctionnalité expérimentale. Activez cette option lorsqu'un seul job synchronise de nombreuses tables. Remarque Ce paramètre est pris en charge uniquement à partir de la version VVR 11.4.

### **Réutiliser un catalogue existant** À partir de VVR 11.5, vous pouvez référencer directement un catalogue MySQL intégré créé sur la page Data Management dans une tâche d'ingestion de données Flink CDC. Cela réduit les efforts manuels liés à l'écriture des propriétés de connexion. ```yaml source: type: mysql using.built-in-catalog: mysql_rds_catalog ``` Actuellement, les tâches d'ingestion de données prennent en charge la réutilisation automatique des paramètres de catalogue MySQL suivants :

hostname

port

username

password

catalog.table.metadata-columns

catalog.table.treat-tinyint1-as-boolean

Si vous souhaitez remplacer l'un de ces paramètres réutilisés automatiquement, vous pouvez spécifier explicitement le paramètre YAML correspondant. Le paramètre spécifié explicitement est prioritaire. ### **Mappage des types** Le tableau suivant présente le mappage des types de données pour l'ingestion de données.

Type de champ MySQL CDC	Type de champ CDC
TINYINT(n)	TINYINT
SMALLINT	SMALLINT
TINYINT UNSIGNED
TINYINT UNSIGNED ZEROFILL
YEAR	INT
INT
MEDIUMINT
MEDIUMINT UNSIGNED
MEDIUMINT UNSIGNED ZEROFILL
SMALLINT UNSIGNED
SMALLINT UNSIGNED ZEROFILL
BIGINT	BIGINT
INT UNSIGNED
INT UNSIGNED ZEROFILL
BIGINT UNSIGNED	DECIMAL(20, 0)
BIGINT UNSIGNED ZEROFILL
SERIAL
FLOAT [UNSIGNED] [ZEROFILL]	FLOAT
DOUBLE [UNSIGNED] [ZEROFILL]	DOUBLE
DOUBLE PRECISION [UNSIGNED] [ZEROFILL]
REAL [UNSIGNED] [ZEROFILL]
NUMERIC(p, s) [UNSIGNED] [ZEROFILL] et p <= 38	DECIMAL(p, s)
DECIMAL(p, s) [UNSIGNED] [ZEROFILL] et p <= 38
FIXED(p, s) [UNSIGNED] [ZEROFILL] et p <= 38
BOOLEAN	BOOLEAN
BIT(1)
TINYINT(1)
DATE	DATE
TIME [(p)]	TIME [(p)]
DATETIME [(p)]	TIMESTAMP [(p)]
TIMESTAMP [(p)]	Le mappage dépend de la valeur du paramètre `treat-timestamp-as-datetime-enabled` : `true` : TIMESTAMP[(p)] `false` : TIMESTAMP_LTZ[(p)]
CHAR(n)	CHAR(n)
VARCHAR(n)	VARCHAR(n)
BIT(n)	BINARY(⌈(n + 7) / 8⌉)
BINARY(n)	BINARY(n)
VARBINARY(N)	VARBINARY(N)
NUMERIC(p, s) [UNSIGNED] [ZEROFILL] et 38 < p <= 65	STRING Remarque Dans MySQL, le type de données décimal offre une précision allant jusqu'à 65, mais dans Flink, la précision est limitée à 38. Par conséquent, si vous définissez une colonne décimale avec une précision supérieure à 38, vous devez la mapper vers une chaîne de caractères pour éviter toute perte de précision.
DECIMAL(p, s) [UNSIGNED] [ZEROFILL] et 38 < p <= 65
FIXED(p, s) [UNSIGNED] [ZEROFILL] et 38 < p <= 65
TINYTEXT	STRING
TEXT
MEDIUMTEXT
LONGTEXT
ENUM
JSON	STRING Remarque Le type de données JSON est converti en une chaîne au format JSON dans Flink.
GEOMETRY	STRING Remarque Les types de données spatiales dans MySQL sont convertis en chaînes avec un format JSON fixe. Pour plus d'informations, consultez la section Spatial Data Type Mapping de la documentation MySQL.
POINT
LINESTRING
POLYGON
MULTIPOINT
MULTILINESTRING
MULTIPOLYGON
GEOMETRYCOLLECTION
TINYBLOB	BYTES Remarque Pour le type de données BLOB dans MySQL, seuls les blobs d'une longueur inférieure ou égale à 2 147 483 647 (2**31-1) sont pris en charge.
BLOB
MEDIUMBLOB
LONGBLOB

Définir un ID de serveur pour éviter les conflits de consommation des journaux binaires

Lorsqu'une tâche d'ingestion de données lit les journaux binaires, la source s'enregistre auprès de MySQL en tant que client de réplication en utilisant un server-id. Si plusieurs tâches ou autres clients de réplication utilisent le même ID de serveur, des conflits de consommation des journaux binaires se produisent et les tâches échouent. Tenez compte des points suivants lors de la configuration des ID de serveur :

Par défaut, un server-id est une valeur unique aléatoire comprise entre 5400 et 6400. Si plusieurs tâches utilisent les valeurs par défaut, des conflits peuvent survenir. Nous vous recommandons de configurer explicitement des ID non chevauchants pour les tâches.
Si le parallélisme de la source est supérieur à 1, vous devez configurer une plage d'ID de serveur. Le nombre d'ID disponibles dans la plage doit être supérieur ou égal au parallélisme, et chaque lecteur parallèle utilise un ID différent.

Exemple : si le parallélisme de la source est de 4, configurez une plage contenant quatre ID.

source:
  type: mysql
  name: MySQL Source
  hostname: <hostname>
  port: 3306
  username: <username>
  password: <password>
  tables: app_db.\.*
  server-id: 5400-5403

sink:
  type: hologres

Si plusieurs tâches lisent à partir de la même instance MySQL, attribuez une plage non chevauchante à chaque tâche. Par exemple, la tâche A utilise 5400-5403 et la tâche B utilise 5404-5407.

Accélérer la lecture des journaux binaires

Lorsque vous utilisez le connecteur MySQL comme source d'ingestion de données, il analyse les fichiers de journaux binaires pour générer divers messages de modification lors de la phase incrémentielle. Les fichiers de journaux binaires enregistrent toutes les modifications de table au format binaire. Vous pouvez accélérer l'analyse des fichiers de journaux binaires de plusieurs manières.

Activez l'analyse parallèle et les filtres d'analyse (cette fonctionnalité nécessite Realtime Compute for Apache Flink avec Ververica Runtime (VVR) 8.0.7 ou une version ultérieure. Elle n'est pas disponible dans l'édition communautaire du connecteur MySQL CDC.)
- Activez l'option scan.only.deserialize.captured.tables.changelog.enabled pour analyser les événements de modification uniquement pour les tables spécifiées.
- Activez l'option scan.parallel-deserialize-changelog.enabled afin d'utiliser plusieurs threads pour analyser le fichier de journal binaire et transmettre les événements à la file d'attente du consommateur dans l'ordre. Lorsque vous activez cette option, vous devez généralement augmenter également le TaskManager CPU.
Optimisez les paramètres Debezium
```
debezium.max.queue.size: 162580
debezium.max.batch.size: 40960
debezium.poll.interval.ms: 50
```
- debezium.max.queue.size : nombre maximal d'enregistrements que la file d'attente bloquante peut contenir. Lorsque Debezium lit un flux d'événements depuis la base de données, il place les événements dans une file d'attente bloquante avant de les écrire en aval. La valeur par défaut est 8192.
- debezium.max.batch.size : nombre maximal d'événements traités par le connecteur à chaque itération. La valeur par défaut est 2048.
- debezium.poll.interval.ms : délai en millisecondes que le connecteur doit attendre avant de demander de nouveaux événements de modification. La valeur par défaut est de 1000 millisecondes, soit 1 seconde.

Exemple d'utilisation :

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: ${mysql.source.table}
  server-id: 7601-7604
  # Debezium configuration
  debezium.max.queue.size: 162580
  debezium.max.batch.size: 40960
  debezium.poll.interval.ms: 50
  # Enable parsing filter
  scan.only.deserialize.captured.tables.changelog.enabled: true

La capacité de consommation des journaux binaires de l'édition Enterprise de MySQL CDC est de 85 Mo/s, soit environ le double de celle de la version communautaire open source. Lorsque la vitesse de génération des fichiers de journaux binaires dépasse 85 Mo/s (soit un fichier de 512 Mo toutes les 6 secondes), la latence du job Flink augmente continuellement. La latence de traitement diminue progressivement après le ralentissement de la vitesse de génération des fichiers de journaux binaires. Si un fichier de journal binaire contient une transaction volumineuse, la latence de traitement peut augmenter temporairement. La latence diminue une fois que le journal de cette transaction a été lu.

Diagnostiquer la latence des données pour optimiser le débit du job

Si vous rencontrez une latence des données pendant la phase incrémentielle, analysez le problème en suivant ces étapes :

Consultez les métriques currentFetchEventTimeLag et currentEmitEventTimeLag sur la page Vue d'ensemble. La métrique currentFetchEventTimeLag représente la latence de lecture des données depuis le journal binaire. La métrique currentEmitEventTimeLag représente la latence de lecture des données pour les tables pertinentes pour le job depuis le journal binaire.

Scénario

Description

currentFetchEventTimeLag est faible, tandis que currentEmitEventTimeLag est élevé et rarement mis à jour.

Une valeur faible de currentFetchEventTimeLag indique que l'extraction du journal binaire depuis la base de données est efficace. Toutefois, le journal binaire contient peu de données pour les tables que le job doit lire. Par conséquent, currentEmitEventTimeLag est rarement mis à jour. Il s'agit d'un comportement attendu.

currentFetchEventTimeLag et currentEmitEventTimeLag sont tous deux élevés.

Cela indique que la table source présente de mauvaises performances de lecture. Vous pouvez passer aux étapes suivantes de cette rubrique pour procéder à l'optimisation.

La contre-pression peut réduire la vitesse à laquelle la source envoie des données aux opérateurs en aval. Vous pouvez observer que sourceIdleTime augmente périodiquement, et que currentFetchEventTimeLag et currentEmitEventTimeLag augmentent continuellement. Pour résoudre ce problème, augmentez le parallélisme du nœud à l'origine de la contre-pression.
Consultez la métrique TM CPU Usage sur la page CPU et la métrique TM GC Time sur la page JVM pour déterminer si les ressources CPU ou mémoire sont insuffisantes. Vous pouvez augmenter les ressources du job pour optimiser les performances de lecture.

Lire les journaux binaires archivés depuis OSS

Lorsque vous utilisez une instance ApsaraDB RDS for MySQL comme source de données, vous pouvez lire les sauvegardes de journaux stockées dans OSS. Si le fichier correspondant à l'horodatage spécifié ou à la position du journal binaire est stocké dans OSS, Flink extrait automatiquement le fichier journal depuis OSS vers le cluster. Si le fichier est stocké localement dans la base de données, Flink bascule automatiquement vers la lecture via une connexion à la base de données. Cette fonctionnalité est disponible uniquement dans Realtime Compute for Apache Flink et n'est pas prise en charge dans l'édition communautaire du connecteur MySQL CDC.

Pour activer la lecture depuis les sauvegardes de journaux OSS, vous devez configurer les paramètres de connexion ApsaraDB RDS for MySQL. Exemple :

source:
  type: mysql
  hostname: <yourHostname>
  port: 3306
  username: <yourUsername>
  password: <yourPassword>
  tables: <yourTables>
  # RDS connection parameters for reading archived binlogs from OSS
  rds.region-id: cn-beijing
  rds.access-key-id: your_access_key_id
  rds.access-key-secret: your_access_key_secret
  rds.db-instance-id: rm-xxxxxxxx # Database instance ID.
  rds.main-db-id: 12345678 # Primary database ID.
  rds.endpoint: rds.aliyuncs.com

Ingestion de données pour la synchronisation de bases de données et de schémas

Pour les jobs qui impliquent uniquement une logique de synchronisation des données, nous vous recommandons de les exécuter en tant que jobs d'ingestion de données. Les jobs d'ingestion de données sont profondément optimisés pour les scénarios d'intégration de données. Pour obtenir des instructions d'utilisation, consultez les rubriques Prise en main de l'ingestion de données Flink CDC et Développer un job d'ingestion de données Flink CDC (aperçu public).

Le code suivant montre un exemple de job d'ingestion de données qui synchronise l'intégralité de la base de données app_db de MySQL vers Hologres, y compris les modifications de schéma ultérieures provenant de la base de données amont :

source:
  type: mysql
  hostname: <hostname>
  port: 3306
  username: ${secret_values.mysqlusername}
  password: ${secret_values.mysqlpassword}
  tables: app_db.\.*
  server-id: 5400-5404

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <endpoint>
  dbname: <database-name>
  username: ${secret_values.holousername}
  password: ${secret_values.holopassword}

pipeline:
  name: Sync MySQL Database to Hologres

Découverte de nouvelles tables dans l'ingestion de données

Le connecteur d'ingestion de données MySQL propose des options de configuration pour prendre en charge la découverte de nouvelles tables dans deux scénarios différents.

Paramètre	Description	Notes
`scan.newly-added-table.enabled`	Lorsqu'un job redémarre à partir d'un point de contrôle, cette option synchronise les tables qui n'ont pas été découvertes lors du démarrage précédent. Elle lit à la fois les données instantanées et incrémentielles de ces nouvelles tables.	Cette option est prise en charge uniquement lorsque `scan.startup.mode` est défini sur `initial`. Elle n'a aucun effet dans les autres modes de démarrage.
`scan.binlog.newly-added-table.enabled`	Pendant la phase incrémentielle, cette option synchronise automatiquement les données des tables nouvellement découvertes.	Nous vous recommandons d'activer cette option lors du premier démarrage du job. Le job analyse automatiquement les instructions DDL `CREATE TABLE` et synchronise les données en aval. Si vous activez cette option et redémarrez le job après la création de la table de base de données, cela peut entraîner une perte de données. En mode de démarrage `initial`, les opérations DDL ne sont pas synchronisées en aval tant que la phase d'instantané n'est pas terminée. Les tables créées pendant la phase d'instantané ne peuvent pas être synchronisées automatiquement, même si `scan.binlog.newly-added-table.enabled` est activé.

Important

Pendant la phase de lecture complète, l'enregistrement d'un point de sauvegarde, suivi de l'ajout ou de la suppression de tables sources avant le redémarrage à partir du point de sauvegarde, n'est pas pris en charge. Cette opération empêche le job de lire correctement les données.
N'activez pas simultanément scan.newly-added-table.enabled et scan.binlog.newly-added-table.enabled. L'activation des deux options peut entraîner une duplication des données.