Realtime Compute for Apache Flink:PostgreSQL CDC

Le connecteur Postgres CDC prend en charge les fonctionnalités suivantes :

Emplacements de réplication

Les tâches Flink PostgreSQL CDC utilisent des emplacements de réplication pour empêcher la purge prématurée du WAL et garantir la cohérence des données. Une mauvaise gestion des emplacements peut entraîner une utilisation excessive du disque ou des retards de lecture. Bonnes pratiques :

• Nettoyez rapidement les emplacements inutilisés

  • Flink ne supprime pas automatiquement les emplacements de réplication après l'arrêt d'une tâche ou un redémarrage sans état, afin d'éviter la perte de données WAL.

  • Si une tâche ne redémarre pas, supprimez manuellement son emplacement de réplication pour libérer de l'espace disque.

    Remarque

    Gestion du cycle de vie : Traitez les emplacements de réplication comme des ressources au niveau de la tâche et gérez-les conjointement avec le démarrage et l'arrêt des tâches.

• Évitez de réutiliser d'anciens emplacements

  • Utilisez toujours un nouveau nom d'emplacement. La réutilisation d'un ancien emplacement oblige la tâche à lire les données WAL historiques accumulées au démarrage, ce qui retarde le traitement des nouvelles données.

  • PostgreSQL requiert un emplacement par connexion. Chaque tâche doit utiliser un nom d'emplacement unique.

    Remarque

    Convention de nommage : Lorsque vous personnalisez slot.name, évitez les noms avec des suffixes numériques tels que my_slot_1 pour prévenir les conflits avec les emplacements temporaires.

• Comportement des emplacements avec les instantanés incrémentiels activés

  • Prérequis : Les points de contrôle (checkpoints) doivent être activés et la table source doit avoir une clé primaire définie.

  • Règles de création des emplacements :

    • Instantané incrémentiel désactivé : Seul un parallélisme de 1 est pris en charge. Un emplacement global est utilisé.

    • Instantané incrémentiel activé :

      • Phase d'instantané : Chaque sous-tâche source concurrente crée un emplacement temporaire. Le format de nommage est ${slot.name}_${task_id}.

      • Phase incrémentielle : Tous les emplacements temporaires sont automatiquement récupérés. Seul un emplacement global est conservé.

  • Nombre maximal d'emplacements : Parallélisme de la source + 1 (pendant la phase d'instantané)

• Ressources et performances

  • Si les emplacements disponibles ou l'espace disque sont limités, réduisez le parallélisme de l'instantané pour utiliser moins d'emplacements temporaires. Cela réduit la vitesse de lecture de l'instantané.

  • Si la table de destination en aval supporte les écritures idempotentes, définissez scan.incremental.snapshot.backfill.skip = true pour ignorer le remplissage arrière du WAL pendant la phase d'instantané et accélérer le démarrage.

    Cela fournit uniquement une sémantique at-least-once et ne convient pas aux calculs avec état (agrégations ou jointures de recherche), car les modifications historiques requises peuvent être perdues.

• Lorsque les instantanés incrémentiels sont désactivés, les points de contrôle ne sont pas pris en charge pendant la phase d'instantané.

  Configuration pour éviter les délais d'expiration pendant la phase d'instantané

  Lorsque les instantanés incrémentiels sont désactivés, les points de contrôle pendant la phase d'instantané peuvent provoquer des basculements dus à des délais d'expiration. Configurez ces paramètres dans Other Configuration (Configuration des paramètres d'exécution personnalisés) :

  execution.checkpointing.interval: 10min
  execution.checkpointing.tolerable-failed-checkpoints: 100
  restart-strategy: fixed-delay
  restart-strategy.fixed-delay.attempts: 2147483647

  Paramètres :

  Paramètre	Description	Remarques
  execution.checkpointing.interval	L'intervalle entre les points de contrôle.	L'unité est une valeur de durée, telle que 10min ou 30s.
  execution.checkpointing.tolerable-failed-checkpoints	Le nombre d'échecs de point de contrôle tolérés avant l'échec de la tâche.	Le produit de ce paramètre et de l'intervalle de planification des points de contrôle correspond au temps de lecture d'instantané autorisé. Remarque Si la table est très volumineuse, attribuez une valeur plus élevée à ce paramètre.
  restart-strategy	La stratégie de redémarrage de la tâche.	Valeurs valides : fixed-delay : stratégie de redémarrage à délai fixe. failure-rate : stratégie de redémarrage basée sur le taux d'échec. exponential-delay : stratégie de redémarrage à délai exponentiel. Stratégies de redémarrage.
  restart-strategy.fixed-delay.attempts	Nombre maximal de tentatives de redémarrage pour la stratégie de redémarrage fixed-delay.	–

Réutilisation de l'abonnement Postgres

Le connecteur Postgres CDC s'appuie sur une publication pour déterminer quelles modifications de tables sont poussées vers un emplacement. Si plusieurs tâches partagent la même publication, leurs configurations seront écrasées.

Cause

La valeur par défaut de publication.autocreate.mode est filtered ; elle n'inclut que les tables présentes dans la configuration du connecteur. Cela modifie la publication au démarrage de la tâche, ce qui peut affecter d'autres tâches.

Solution

1. Créez une publication dans PostgreSQL qui inclut toutes les tables surveillées, ou créez une publication distincte par tâche.

   -- Create a publication named my_flink_pub that includes all tables (or specified tables, creating one publication per job)
   CREATE PUBLICATION my_flink_pub FOR TABLE table_a, table_b;
   -- Or more simply, include all tables in the database
   CREATE PUBLICATION my_flink_pub FOR ALL TABLES;

   Remarque

   L'abonnement à toutes les tables n'est pas recommandé pour les bases de données volumineuses en raison d'une consommation excessive de bande passante et de CPU sur le cluster Flink.

2. Ajoutez les configurations Flink suivantes :

   • debezium.publication.name = 'my_flink_pub' (Spécifie le nom de la publication)

   • debezium.publication.autocreate.mode = 'disabled' (Empêche Flink de tenter de créer ou de modifier la publication au démarrage)

Cela assure une isolation complète et empêche les nouvelles tâches d'affecter les tâches existantes.

Syntaxe

CREATE TABLE postgrescdc_source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>',
  'decoding.plugin.name'= 'pgoutput',
  'scan.incremental.snapshot.enabled' = 'true',
  -- Skipping backfill can speed up reads and reduce resource usage, but it may cause data duplication. Enable this if the downstream sink is idempotent.
  'scan.incremental.snapshot.backfill.skip' = 'false',
  -- In a production environment, set this to 'filtered' or 'disabled' and manage the publication manually instead of through Flink.
  'debezium-publication.autocreate.mode' = 'disabled'
  -- If you have multiple sources, configure a different publication for each source.
  --'debezium.publication.name' = 'my_flink_pub'
);

Options du connecteur

Mappages de types

Exemple

CREATE TABLE source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>'
);

SELECT * FROM source;

Références

• Connecteurs pris en charge

• PolarDB for PostgreSQL (compatible Oracle)

• MySQL