Realtime Compute for Apache Flink:Connecteur SQL MySQL

Le connecteur MySQL offre les fonctionnalités suivantes.

• Traitement unifié par lots et en flux qui permet de lire à la fois les données complètes et incrémentielles, éliminant ainsi la nécessité de maintenir deux processus distincts.

• Lecture simultanée des données complètes pour une mise à l'échelle horizontale des performances.

• Basculement transparent de la lecture des données complètes vers la lecture des données incrémentielles et mise à l'échelle automatique vers le bas pour économiser les ressources de calcul.

• Transfert de données reprenable pendant la phase de lecture des données complètes pour une stabilité accrue.

• Lecture sans verrou des données complètes, qui n'affecte pas les services en ligne.

• Prise en charge de la lecture des journaux de sauvegarde d'ApsaraDB RDS for MySQL.

• Analyse parallèle des fichiers de journaux binaires pour réduire la latence de lecture.

SQL

Vous pouvez utiliser le connecteur MySQL dans les jobs SQL en tant que table source, table de dimension ou table de destination.

Syntaxe

CREATE TEMPORARY TABLE mysqlcdc_source (
   order_id INT,
   order_date TIMESTAMP(0),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   order_status BOOLEAN,
   PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = '<yourHostname>',
  'port' = '3306',
  'username' = '<yourUsername>',
  'password' = '<yourPassword>',
  'database-name' = '<yourDatabaseName>',
  'table-name' = '<yourTableName>'
);

Paramètres WITH

• Général

  Paramètre	Description	Obligatoire	Type de données	Valeur par défaut	Notes
  connector	Le type de table.	Oui	STRING	Aucune	Lorsqu'il est utilisé comme table source, vous pouvez définir ce paramètre sur mysql-cdc ou mysql. Ces deux valeurs sont équivalentes. Lorsqu'il est utilisé comme table de dimension ou table de destination, la valeur doit être mysql.
  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 (Virtual Private Cloud). 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 les rubriques 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.
  database-name	Le nom de la base de données MySQL.	Oui	STRING	Aucune	Lorsqu'une base de données est utilisée comme table source, vous pouvez utiliser une expression régulière pour le nom de la base de données afin de lire les données provenant de plusieurs bases. Lorsque vous utilisez des expressions régulières, n'utilisez pas les symboles ^ et $ pour faire correspondre le début et la fin de la chaîne. Pour plus d'informations, consultez les notes relatives au paramètre table-name.
  table-name	Le nom de la table MySQL.	Oui	STRING	Aucune	Vous pouvez utiliser une expression régulière pour le nom de la table source afin de lire les données provenant de plusieurs tables. Lorsque vous lisez des données depuis plusieurs tables MySQL, soumettez plusieurs instructions CTAS dans le cadre d'une seule tâche. Cela évite d'activer plusieurs écouteurs de journaux binaires et améliore les performances et l'efficacité. Pour plus d'informations, consultez la rubrique Plusieurs instructions CTAS : Soumission en tant que tâche unique. Lorsque vous utilisez des expressions régulières, n'utilisez pas les symboles ^ et $ pour faire correspondre le début et la fin de la chaîne. Pour plus d'informations, consultez la note suivante. Remarque Lorsqu'une table source MySQL CDC fait correspondre des noms de table à l'aide d'une expression régulière, elle concatène les paramètres database-name et table-name que vous avez spécifiés avec la chaîne \\. pour former une expression régulière de chemin complet. Avant la version VVR 8.0.1, le caractère . était utilisé. Le connecteur utilise ensuite cette expression régulière pour faire correspondre les noms qualifiés complets des tables dans la base de données MySQL. Par exemple, si vous définissez 'database-name'='db_.' et 'table-name'='tb_.+', le connecteur utilise l'expression régulière db_.\\.tb_.+ pour faire correspondre les noms de table qualifiés complets et déterminer quelles tables lire. Avant la version VVR 8.0.1, l'expression régulière était db_.*.tb_.+.
  port	Le numéro de port du service de base de données MySQL.	Non	INTEGER	3306	Aucune.

• Table source uniquement

  Paramètre	Description	Obligatoire	Type de données	Valeur par défaut	Notes
  server-id	Identifiant numérique du client de base de données.	Non	STRING	Une valeur aléatoire comprise entre 5400 et 6400 est générée.	Cet identifiant doit être globalement unique au sein du cluster MySQL. Définissez un identifiant différent pour chaque tâche qui se connecte à la même base de données. Ce paramètre prend également en charge le format de plage d'identifiants, par exemple 5400-5408. Lorsque la lecture incrémentielle est activée, la lecture simultanée est prise en charge. Dans ce cas, définissez une plage d'identifiants afin que chaque lecteur simultané utilise un identifiant différent. Pour plus d'informations, consultez Utiliser l'ID de serveur.
  scan.incremental.snapshot.enabled	Indique s'il faut activer les snapshots incrémentiels.	Non	BOOLEAN	true	Les snapshots incrémentiels sont activés par défaut. Le snapshot incrémentiel est un nouveau mécanisme de lecture des snapshots complets de données. Par rapport à l'ancienne méthode de lecture des snapshots, les snapshots incrémentiels présentent de nombreux avantages, notamment : La source peut lire les données complètes en parallèle. La source prend en charge les points de contrôle au niveau des fragments lors de la lecture des données complètes. La source n'a pas besoin d'acquérir un verrou de lecture global (FLUSH TABLES WITH read lock) lors de la lecture des données complètes. Si vous souhaitez que la source prenne en charge la lecture simultanée, chaque lecteur simultané doit disposer d'un ID de serveur unique. Par conséquent, server-id doit être une plage, telle que 5400-6400, et la taille de la plage doit être supérieure ou égale au niveau de simultanéité. Remarque Cet élément de configuration est supprimé dans Ververica Runtime (VVR) 11.1 et versions ultérieures.
  scan.incremental.snapshot.chunk.size	Taille de chaque fragment, exprimée en nombre de lignes.	Non	INTEGER	8096	Lorsque la lecture par snapshot incrémentiel est activée, la table est divisée 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 de dépassement de mémoire (OOM) et une diminution du débit global. Vous devez donc trouver un compromis et définir une taille de fragment raisonnable.
  scan.snapshot.fetch.size	Nombre maximal d'enregistrements à extraire à la fois lors de la lecture des données complètes d'une table.	Non	INTEGER	1024	Aucun.
  scan.startup.mode	Mode de démarrage pour la consommation de données.	Non	STRING	initial	Valeurs valides : initial (par défaut) : Lors du premier démarrage ou 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 : Lors du premier démarrage ou 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 effectué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 indiqué par scan.startup.timestamp-millis en millisecondes. Important Lors de l'utilisation du mode de démarrage earliest-offset, specific-offset ou timestamp, assurez-vous 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. Cela permet d'éviter les erreurs dues à des incompatibilités de schéma.
  scan.startup.specific-offset.file	Nom du fichier du journal binaire pour le décalage de départ lors de l'utilisation du mode de démarrage specific-offset.	Non	STRING	Aucun	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	Décalage au sein du fichier de journal binaire spécifié pour le décalage de départ lors de l'utilisation du mode de démarrage specific-offset.	Non	INTEGER	Aucun	Lorsque vous utilisez ce paramètre, vous devez définir scan.startup.mode sur specific-offset.
  scan.startup.specific-offset.gtid-set	Ensemble GTID pour le décalage de départ lors de l'utilisation du mode de démarrage specific-offset.	Non	STRING	Aucun	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	Horodatage en millisecondes pour le décalage de départ lors de l'utilisation du mode de démarrage timestamp.	Non	LONG	Aucun	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	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 de MySQL en type STRING. Pour plus d'informations, consultez Valeurs temporelles Debezium.
  debezium.min.row.count.to.stream.results	Lorsque le nombre de lignes d'une table est supérieur à cette valeur, le mode de lecture par lots est utilisé.	Non	INTEGER	1000	Flink lit les données d'une table source MySQL de l'une des manières suivantes : Lecture complète : lit directement toutes les données de la table en mémoire. Cette méthode est rapide, mais consomme une quantité de mémoire correspondante. Si la table source est très volumineuse, il existe un risque d'erreurs OOM. Lecture par lots : lit les données en plusieurs lots, avec un certain nombre de lignes par lot, jusqu'à ce que toutes les données soient lues. Cette méthode évite les risques d'erreur OOM lors de la lecture de tables volumineuses, mais est relativement lente.
  connect.timeout	Délai maximal d'attente avant expiration de la connexion au serveur de base de données MySQL avant nouvelle tentative.	Non	DURATION	30s	Aucun.

  connect.max-retries	Nombre maximal de tentatives après un échec de connexion au service de base de données MySQL.	Non	INTEGER	3	Aucun.
  connection.pool.size	Taille du pool de connexions à la base de données.	Non	INTEGER	20	Le pool de connexions permet de réutiliser les connexions, ce qui réduit le nombre de connexions à la base de données.
  jdbc.properties.*	Paramètres de connexion personnalisés dans l'URL JDBC.	Non	STRING	Aucun	Vous pouvez transmettre des paramètres de connexion personnalisés. Par exemple, pour ne pas utiliser le protocole SSL, configurez '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 permettant à Debezium de lire les journaux binaires.	Non	STRING	Aucun	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 les paramètres Debezium de manière arbitraire. Cela pourrait entraîner une lecture incorrecte des données par le connecteur. Par exemple, il n'est pas autorisé de configurer le paramètre debezium.binlog.buffer.size.
  heartbeat.interval	Intervalle auquel la source fait avancer le décalage du journal binaire à l'aide d'événements de pulsation (heartbeat).	Non	DURATION	30 s	Les événements de pulsation permettent de faire avancer le décalage du journal binaire dans la source. Cette fonctionnalité est particulièrement utile pour les tables MySQL mises à jour rarement. Pour ces tables, le décalage du journal binaire ne peut pas avancer automatiquement. Les événements de pulsation font progresser ce décalage, évitant ainsi les problèmes liés à son expiration. Un décalage expiré peut provoquer l'échec irréversible du job, nécessitant un redémarrage sans état.
  scan.incremental.snapshot.chunk.key-column	Spécifie une colonne à utiliser comme colonne de fractionnement pour le sharding pendant la phase de snapshot.	Consultez la colonne Notes.	STRING	Aucun	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 de la clé primaire peut être sélectionnée.
  rds.region-id	ID de région de l'instance ApsaraDB RDS for MySQL d'Alibaba Cloud.	Requis lors de l'utilisation de la fonctionnalité de lecture des journaux archivés depuis OSS.	STRING	Aucun	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'augmente pas de façon monotone comme les décalages de fichiers de journaux binaires, localiser un GTID dans un fichier nécessite de télécharger et d'analyser tous les journaux archivés depuis OSS. Ce processus consomme beaucoup de ressources et prend du temps, rendant impossibles les fonctionnalités reposant 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 MySQL primaire/secondaire reposent sur les GTID. Évaluez soigneusement cette fonctionnalité avant utilisation.
  rds.access-key-id	AccessKey ID du compte ApsaraDB RDS for MySQL d'Alibaba Cloud.	Requis lors de l'utilisation de la fonctionnalité de lecture des journaux archivés depuis OSS.	STRING	Aucun	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	AccessKey secret du compte ApsaraDB RDS for MySQL d'Alibaba Cloud.	Requis lors de l'utilisation de la fonctionnalité de lecture des journaux archivés depuis OSS.	STRING	Aucun	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	ID de l'instance ApsaraDB RDS for MySQL d'Alibaba Cloud.	Requis lors de l'utilisation de la fonctionnalité de lecture des journaux archivés depuis OSS.	STRING	Aucun	Aucun.
  rds.main-db-id	Numéro de la base de données principale de l'instance ApsaraDB RDS for MySQL d'Alibaba Cloud.	Non	STRING	Aucun	Pour savoir comment obtenir le numéro de la base de données principale, consultez Sauvegarde des journaux ApsaraDB RDS for MySQL. Pris en charge uniquement dans VVR 8.0.7 et versions ultérieures. 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	Délai d'expiration pour le téléchargement d'un seul journal archivé depuis OSS.	Non	DURATION	60 s	Aucun.
  rds.endpoint	Endpoint de service permettant d'obtenir les informations sur les journaux binaires OSS.	Non	STRING	Aucun	Pour plus d'informations sur les valeurs valides, consultez Endpoints. Pris en charge uniquement dans VVR 8.0.8 et versions ultérieures.
  scan.incremental.close-idle-reader.enabled	Indique s'il faut fermer les lecteurs inactifs une fois le snapshot terminé.	Non	BOOLEAN	false	Pris en charge uniquement dans VVR 8.0.1 et versions ultérieures. Pour que cette configuration prenne effet, vous devez définir execution.checkpointing.checkpoints-after-tasks-finish.enabled sur true.
  scan.read-changelog-as-append-only.enabled	Indique s'il faut convertir le flux de données changelog en un flux de données append-only.	Non	BOOLEAN	false	Valeurs valides : true : Tous les types de messages, y compris INSERT, DELETE, UPDATE_BEFORE et UPDATE_AFTER, sont convertis en messages INSERT. Activez cette option uniquement dans des scénarios spécifiques, par exemple lorsque vous devez conserver les messages de suppression de la table amont. false (par défaut) : Tous les types de messages sont transmis en aval tels quels. Remarque Pris en charge uniquement dans VVR 8.0.8 et versions ultérieures.

  scan.only.deserialize.captured.tables.changelog.enabled	En phase incrémentielle, 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. Remarque Pris en charge uniquement à partir de la version VVR 8.0.7. Si vous utilisez une version VVR 8.0.8 ou antérieure, vous devez modifier le nom du paramètre en debezium.scan.only.deserialize.captured.tables.changelog.enable.
  scan.parse.online.schema.changes.enabled	En phase incrémentielle, 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 snapshot du job Flink pour permettre la récupération. Remarque Pris en charge uniquement à partir de la version VVR 11.1.
  scan.incremental.snapshot.backfill.skip	Indique s'il faut ignorer le remplissage rétroactif (backfill) pendant la phase de lecture du snapshot.	Non	BOOLEAN	false	Valeurs valides : true : Ignore le remplissage rétroactif pendant la phase de lecture du snapshot. false (par défaut) : N'ignore pas le remplissage rétroactif pendant la phase de lecture du snapshot. Le remplissage rétroactif s'applique uniquement lors de l'interrogation du snapshot 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é, l'interrogation du snapshot de chaque fragment lit les dernières données de la table à cet instant précis ; les mises à jour intervenant sur un fragment après sa lecture ne sont pas fusionnées pendant la phase de lecture complète et sont lues depuis le Binlog une fois la phase incrémentielle engagée. Par exemple, une mise à jour du fragment5 qui survient pendant la prise du snapshot du fragment5 est directement reflétée dans le snapshot du fragment5 ; si le fragment5 est mis à jour après que le lecteur soit passé au fragment80, la mise à jour sera appliquée ultérieurement depuis le Binlog pendant la phase incrémentielle. Important Lorsque cette option est activée, les modifications survenant pendant ou après l'analyse d'un fragment sont toujours transmises depuis le Binlog en phase incrémentielle 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 supporte les écritures idempotentes par clé primaire. Remarque Pris en charge uniquement à partir de la version VVR 11.1.
  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 du snapshot.	Non	BOOELEAN	false	Valeurs valides : true : Distribue en priorité les fragments non bornés pendant la phase de lecture du snapshot. false (par défaut) : Ne distribue pas en priorité les fragments non bornés pendant la phase de lecture du snapshot. Il s'agit d'une fonctionnalité expérimentale. Son activation peut réduire le risque d'erreurs OOM (Out Of Memory) sur le TaskManager lors de la synchronisation du dernier fragment pendant la phase de snapshot. 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 en lecture/écriture 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	Aucun	Ce paramètre s'applique aux scénarios nécessitant de limiter la lecture des données. Cette limite est effective tant en phase complète qu'en phase incrémentielle. 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. En phase de lecture complète des données, il est généralement nécessaire de réduire le nombre de lignes lues dans chaque lot. Pour ce faire, vous pouvez diminuer la valeur du paramètre scan.incremental.snapshot.chunk.size. Remarque Pris en charge uniquement à partir de la version VVR 11.5.
  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 poursuivre son exécution.	Non	BOOLEAN	false	Avant d'activer ce paramètre, vous devez vous assurer que l'offset de démarrage du job n'a pas expiré. Si le job démarre à partir d'un offset GTID effacé ou expiré, le moteur ignorera continuellement 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.

• Paramètres spécifiques aux tables de dimension

  Paramètre	Description	Obligatoire	Type de données	Valeur par défaut	Notes
  url	L'URL JDBC MySQL.	Non	STRING	Aucune	Le format de l'URL est : jdbc:mysql://<endpoint>:<port>/<database_name>.
  lookup.max-retries	Le nombre maximal de tentatives après un échec de lecture des données.	Non	INTEGER	3	Pris en charge uniquement dans VVR 6.0.7 et versions ultérieures.
  lookup.cache.strategy	La politique de cache.	Non	STRING	Aucune	Les politiques de cache prises en charge sont None, LRU et ALL. Pour plus d'informations sur les valeurs, consultez Instructions JOIN pour les tables de dimension. Remarque Lorsque vous utilisez la politique de cache LRU, vous devez également configurer le paramètre lookup.cache.max-rows.
  lookup.cache.max-rows	Le nombre maximal de lignes mises en cache.	Non	INTEGER	100000	Si vous sélectionnez la politique de cache LRU, vous devez définir la taille du cache. Si vous sélectionnez la politique de cache ALL, il n'est pas nécessaire de définir la taille du cache.
  lookup.cache.ttl	La durée de vie (TTL) du cache.	Non	DURATION	10 s	La configuration de lookup.cache.ttl dépend de lookup.cache.strategy : Si lookup.cache.strategy est défini sur None, vous n'avez pas besoin de configurer lookup.cache.ttl. Cela signifie que le cache n'expire pas. Si lookup.cache.strategy est défini sur LRU, lookup.cache.ttl correspond à la durée de vie du cache. Par défaut, le cache n'expire pas. Si lookup.cache.strategy est défini sur ALL, lookup.cache.ttl correspond au temps de chargement du cache. Par défaut, le cache n'est pas rechargé. Utilisez un format temporel, tel que 1min ou 10s.
  lookup.max-join-rows	Le nombre maximal de résultats renvoyés lorsqu'un enregistrement de la table principale correspond à des enregistrements de la table de dimension.	Non	INTEGER	1024	Aucune.
  lookup.filter-push-down.enabled	Indique s'il faut activer le pushdown de filtre pour la table de dimension.	Non	BOOLEAN	false	Valeurs valides : true : Active le pushdown de filtre pour la table de dimension. Lors du chargement des données depuis la table de base de données MySQL, la table de dimension filtre les données à l'avance en fonction des conditions définies dans la tâche SQL. false (par défaut) : Désactive le pushdown de filtre pour la table de dimension. Lors du chargement des données depuis la table de base de données MySQL, la table de dimension charge toutes les données. Remarque Pris en charge uniquement dans VVR 8.0.7 et versions ultérieures. Important Le pushdown de table de dimension ne doit être activé que lorsqu'une table Flink est utilisée comme table de dimension. Les tables source MySQL ne prennent pas en charge l'activation du pushdown de filtre. Si une table Flink est utilisée à la fois comme table source et comme table de dimension, et que le pushdown de filtre est activé pour la table de dimension, vous devez explicitement définir cet élément de configuration sur false pour la table source à l'aide de SQL Hints. Sinon, la tâche peut s'exécuter de manière anormale.

• Uniquement pour les tables sink

  Paramètre	Description	Obligatoire	Type de données	Valeur par défaut	Notes
  url	L'URL JDBC MySQL.	Non	STRING	Aucune	Le format de l'URL est : jdbc:mysql://<endpoint>:<port>/<database_name>.
  sink.max-retries	Nombre maximal de tentatives après un échec d'écriture des données.	Non	INTEGER	3	Aucune.
  sink.buffer-flush.batch-size	Nombre de lignes dans une écriture par lot unique.	Non	INTEGER	4096	Aucune.
  sink.buffer-flush.max-rows	Nombre de lignes de données mises en cache en mémoire.	Non	INTEGER	10000	Ce paramètre ne prend effet que si une clé primaire est spécifiée.
  sink.buffer-flush.interval	Intervalle de vidage du cache. Si les données du cache ne remplissent pas les conditions de sortie après le temps d'attente spécifié, le système sort automatiquement toutes les données du cache.	Non	DURATION	1s	Aucune.
  sink.ignore-delete	Indique s'il faut ignorer les opérations de suppression (DELETE) des données.	Non	BOOLEAN	false	Lorsque le flux généré par Flink SQL inclut des enregistrements delete ou update-before, des incohérences de données peuvent survenir si plusieurs tâches de sortie mettent à jour simultanément différents champs de la même table. Par exemple, après la suppression d'un enregistrement, une autre tâche met à jour uniquement certains champs. Les champs non mis à jour deviennent alors null ou prennent leurs valeurs par défaut, ce qui entraîne des erreurs de données. En définissant sink.ignore-delete sur true, vous pouvez ignorer les opérations DELETE et UPDATE_BEFORE en amont pour éviter ces problèmes. Remarque UPDATE_BEFORE fait partie du mécanisme de rétractation de Flink, utilisé pour « rétracter » l'ancienne valeur lors d'une opération de mise à jour. Lorsque ignoreDelete = true, tous les enregistrements de type DELETE et UPDATE_BEFORE sont ignorés. Seuls les enregistrements INSERT et UPDATE_AFTER sont traités.
  sink.ignore-delete-mode	Stratégie de gestion des enregistrements de type delete après l'ignorance des opérations DELETE.	Non	STRING	ALL	Valeurs valides : ALL : Ignore les enregistrements -D et -U. REAL_DELETE : Ignore uniquement les enregistrements -D. UPDATE_BEFORE : Ignore uniquement les enregistrements -U. Remarque Cette option est prise en charge uniquement dans le moteur Realtime Compute VVR 11.8 et versions ultérieures. Effectif uniquement lorsque sink.ignore-delete=true. Une configuration isolée de ce paramètre entraîne une erreur.
  sink.ignore-null-when-update	Lors de la mise à jour des données, indique s'il faut mettre à jour le champ correspondant avec la valeur null ou ignorer la mise à jour de ce champ si la valeur du champ de données entrant est null.	Non	BOOLEAN	false	Valeurs valides : true : Ne met pas à jour le champ. Ce paramètre ne peut être défini sur true que si une clé primaire est configurée pour la table Flink. Lorsque la valeur est true : Pour les versions VVR 8.0.6 et antérieures, la table de destination (sink) ne prend pas en charge l'écriture par lot. Pour les versions VVR 8.0.7 et ultérieures, la table de destination (sink) prend en charge l'écriture par lot. L'écriture par lot peut améliorer considérablement l'efficacité d'écriture et le débit global, mais elle introduit une latence des données et un risque d'erreurs OOM (Out Of Memory). Vous devez donc effectuer un compromis en fonction de votre scénario métier. false : Met à jour le champ avec la valeur null. Remarque Ce paramètre est pris en charge uniquement dans les versions VVR 8.0.5 et ultérieures.

Mappage des types

• Tables source CDC

  Type de champ MySQL CDC	Type de champ Flink
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  TINYINT UNSIGNED ZEROFILL
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  SMALLINT UNSIGNED ZEROFILL
  BIGINT	BIGINT
  INT UNSIGNED
  INT UNSIGNED ZEROFILL
  MEDIUMINT UNSIGNED
  MEDIUMINT 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]	DECIMAL(p, s)
  DECIMAL(p, s) [UNSIGNED] [ZEROFILL]
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]	TIMESTAMP [(p)]
  	TIMESTAMP [(p)] WITH LOCAL TIME ZONE
  CHAR(n)	STRING
  VARCHAR(n)
  TEXT
  BINARY	BYTES
  VARBINARY
  BLOB

  Important

  N'utilisez pas le type TINYINT(1) dans MySQL pour stocker des valeurs autres que 0 et 1. Lorsque la propriété property-version=0, la table source MySQL CDC mappe par défaut TINYINT(1) au type BOOLEAN dans Flink. Cela peut entraîner des inexactitudes dans les données. Pour utiliser le type TINYINT(1) afin de stocker des valeurs autres que 0 et 1, consultez le paramètre de configuration catalog.table.treat-tinyint1-as-boolean.

• Tables de dimension et tables de destination (sink)

  Type de champ MySQL	Type de champ Flink
  TINYINT	TINYINT
  SMALLINT	SMALLINT
  TINYINT UNSIGNED
  INT	INT
  MEDIUMINT
  SMALLINT UNSIGNED
  BIGINT	BIGINT
  INT UNSIGNED
  BIGINT UNSIGNED	DECIMAL(20, 0)
  FLOAT	FLOAT
  DOUBLE	DOUBLE
  DOUBLE PRECISION
  NUMERIC(p, s)	DECIMAL(p, s) Remarque où p <= 38.
  DECIMAL(p, s)
  BOOLEAN	BOOLEAN
  TINYINT(1)
  DATE	DATE
  TIME [(p)]	TIME [(p)] [WITHOUT TIME ZONE]
  DATETIME [(p)]	TIMESTAMP [(p)] [WITHOUT TIME ZONE]
  TIMESTAMP [(p)]
  CHAR(n)	CHAR(n)
  VARCHAR(n)	VARCHAR(n)
  BIT(n)	BINARY(⌈n/8⌉)
  BINARY(n)	BINARY(n)
  VARBINARY(N)	VARBINARY(N)
  TINYTEXT	STRING
  TEXT
  MEDIUMTEXT
  LONGTEXT
  TINYBLOB	BYTES Important Flink prend uniquement en charge les enregistrements de type BLOB MySQL dont la taille est inférieure ou égale à 2 147 483 647 (2^31 - 1) octets.
  BLOB
  MEDIUMBLOB
  LONGBLOB

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    -- 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',  -- Parse only the change events of specified tables.
    ...
)

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

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 :

1. Vérifiez les métriques currentFetchEventTimeLag et currentEmitEventTimeLag sur la page Overview. 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. Cependant, 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 section pour l'optimisation.

2. 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.

3. Vérifiez la métrique TM CPU Usage sur la page CPU et la métrique TM GC Time sur la page JVM pour déterminer s'il existe une insuffisance de ressources CPU ou mémoire. Vous pouvez augmenter les ressources du job pour optimiser les performances de lecture. Vous pouvez également activer les paramètres mini-batch pour améliorer le débit. Pour plus d'informations, consultez Techniques d'optimisation haute performance de Flink SQL.

4. Si un opérateur SinkUpsertMaterializer avec un état volumineux existe dans le job, il peut affecter les performances de lecture. Envisagez d'augmenter le parallélisme du job ou d'éviter l'opérateur SinkUpsertMaterializer. Pour plus d'informations, consultez Éviter l'utilisation de SinkUpsertMaterializer. La suppression de l'opérateur SinkUpsertMaterializer d'un job existant nécessite un redémarrage sans état. En effet, la topologie du job change et un démarrage à partir de l'état existant peut entraîner l'échec du job ou une perte de données.

Optimisation des paramètres de chunk

Les données des chunks et leurs métadonnées sont stockées en mémoire, ce qui peut parfois entraîner des erreurs d'épuisement de la mémoire (OOM). Vous pouvez ajuster les paramètres en fonction du composant qui rencontre l'erreur OOM :

• JobManager

  Le JobManager stocke les métadonnées de tous les chunks. Un nombre excessif de chunks peut provoquer une erreur OOM. Pour résoudre ce problème, augmentez la valeur de scan.incremental.snapshot.chunk.size afin de réduire le nombre de chunks. Vous pouvez également augmenter la mémoire heap du JobManager en définissant jobmanager.memory.heap.size dans votre configuration d'exécution. Pour plus d'informations, consultez Configuration des paramètres Flink.

• TaskManager

  • Le TaskManager lit les données de chaque chunk. Si un chunk contient trop de lignes, une erreur OOM peut se produire. Pour résoudre ce problème, diminuez la valeur de scan.incremental.snapshot.chunk.size afin de réduire le nombre de lignes par chunk. Vous pouvez également augmenter la mémoire heap du TaskManager en augmentant la valeur de TaskManager Memory dans votre configuration d'exécution.

  • Dans VVR 8.0.8 et versions antérieures, le dernier chunk peut contenir une grande quantité de données, ce qui peut provoquer une erreur OOM sur le TaskManager. Nous vous recommandons de mettre à niveau vers VVR 8.0.9 ou une version ultérieure pour éviter ce problème.

  • Pour une table source MySQL CDC avec une clé primaire composite, les chunks sont divisés par défaut en fonction de la première colonne de la clé. Si les données sont fortement asymétriques, avec de nombreuses lignes partageant la même valeur dans cette colonne, le chunk correspondant à cette valeur peut devenir très volumineux et provoquer une erreur OOM sur le TaskManager. Vous pouvez définir scan.incremental.snapshot.chunk.key-column pour spécifier une autre colonne de la clé primaire pour la division.

SET 'table.optimizer.source-merge.enabled' = 'true';

CREATE TABLE mysql_source (...) WITH (
    'connector' = 'mysql-cdc',
    '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',  // The ID of the database instance.
    'rds.main-db-id' = '12345678', // The ID of the primary database.
    'rds.endpoint' = 'rds.aliyuncs.com'
    ...
)