Paramétrez Openflow Connector for PostgreSQL¶

Configurer wal_level¶

Openflow Connector for PostgreSQL exige que wal_level soit défini sur logical.

En fonction de l’endroit où votre serveur PostgreSQL est hébergé, vous pouvez configurer wal_level comme suit :

Créer une publication¶

Openflow Connector for PostgreSQL exige qu’une publication https://www.postgresql.org/docs/current/logical-replication-publication.html#LOGICAL-REPLICATION-PUBLICATION soit créée et configurée dans PostgreSQL avant le début de la réplication. Vous pouvez la créer pour toutes les tables ou un sous-ensemble de tables, ainsi que pour des tables spécifiques comportant uniquement des colonnes déterminées. Assurez-vous que toutes les tables et colonnes que vous prévoyez de faire répliquer sont incluses dans la publication. Vous pouvez également modifier la publication ultérieurement, pendant que le connecteur fonctionne. Pour créer et configurer une publication, procédez comme suit :

1. Connectez-vous en tant qu’utilisateur avec le privilège CREATE sur la base de données et exécutez la requête suivante :

   CREATE PUBLICATION <publication name>; 

   Copy

2. Définissez les tables que l’agent de la base de données pourra voir en utilisant :

ALTER PUBLICATION <publication name> ADD TABLE <table name>; 

Copy

Important

PostgreSQL 15 et les versions ultérieures prennent en charge la configuration des publications pour un sous-ensemble spécifié de colonnes de table. Pour que le connecteur prenne cela en charge correctement, vous devez utiliser les paramètres de filtrage des colonnes afin d’inclure les mêmes colonnes que celles définies dans la publication.

Sans ce paramètre, le connecteur se comportera comme suit :

• Dans la table de destination, les colonnes qui ne sont pas incluses dans le filtre comporteront le suffixe __DELETED. Toutes les données répliquées lors de la phase instantanée seront conservées.

• Une fois que vous avez ajouté de nouvelles colonnes à la publication, la table sera définitivement en état d’échec et vous devrez redémarrer sa réplication. réplication.

Pour plus d’informations, voir ALTER PUBLICATION.

Répliquer un ensemble de tables en temps réel¶

1. Cliquez avec le bouton droit de la souris sur le groupe de processus importé et sélectionnez Parameters.

2. Renseignez les valeurs des paramètres requis comme décrit dans Paramètres de débit.

Répliquer des tables à partir d’un serveur de réplique PostgreSQL¶

Le connecteur peut ingérer des données à partir d’un serveur principal, d’une réplique de substitution active, ou d’un serveur d’abonné utilisant la réplication logique. Avant de configurer le connecteur pour qu’il se connecte à une réplique PostgreSQL, assurez-vous que la réplication entre les nœuds principaux et les nœuds de réplique fonctionne correctement. Lorsque vous enquêtez sur des problèmes liés à des données manquantes dans le connecteur, assurez-vous d’abord que les lignes manquantes sont présentes dans le serveur de réplique utilisé par le connecteur.

Considérations supplémentaires lors de la connexion à une réplique de substitution :

• Seule la connexion à une réplique de substitution active est prise en charge. Notez que les répliques de substitution actives ne peuvent pas accepter de connexions provenant de clients tant qu’elles n’ont pas été promues à une instance primaire.

• La version PostgreSQL du serveur doit être >= 16.

• La publication requise par le connecteur doit être créée sur le serveur principal, et non sur le serveur de substitution. Le serveur de substitution est en lecture seule et ne permet pas de créer de publication.

Si vous vous connectez à une instance de substitution active et que vous voyez Tentative de création de l’emplacement de réplication, cela signifie que “<replication slot>” a expiré. Si vous vous connectez à une instance de substitution, assurez-vous qu’il existe un trafic sur l’instance primaire PostgreSQL. Sinon, l’appel à créer un emplacement de réplication ne renverra jamais de réponse. Si vous voyez une erreur ` dans le bulletin Openflow, ou si le processeur :ui:`Read PostgreSQL CDC Stream ne démarre pas, connectez-vous à l’instance PostgreSQL principale et exécutez la requête suivante :

SELECT pg_log_standby_snapshot(); 

L’erreur se produit lorsqu’il n’y a pas de modifications de données dans le serveur principal. Ainsi, le connecteur peut se bloquer lors de la création d’un emplacement de réplication sur le serveur de réplique. Cela est lié au fait que le serveur de réplique nécessite des informations sur les transactions en cours d’exécution du serveur principal pour pouvoir créer un emplacement de réplication. Les serveurs principaux n’enverront pas les informations lorsqu’ils seront inactifs. La fonction pg_log_standby_snapshot() force le serveur principal à envoyer des informations sur les transactions en cours d’exécution au serveur de réplique.

Supprimez et ajoutez à nouveau une table à la réplication¶

Pour supprimer une table de la réplication, assurez-vous qu’elle est supprimée des paramètres Noms des tables incluses ou Table incluse Regex dans le contexte des paramètres de réplication.

Si vous souhaitez réajuster la table à la réplication ultérieurement, supprimez d’abord la table de destination correspondante dans Snowflake. Ensuite, ajoutez à nouveau la table aux paramètres Noms des tables incluses ou Table incluse Regex. Cela permet de garantir que le processus de réplication démarre à nouveau pour la table.

Cette approche peut également être utilisée pour récupérer un scénario de réplication de table qui a échoué.

Répliquer un sous-ensemble de colonnes dans une table¶

Le connecteur peut filtrer les données répliquées par table sur un sous-ensemble de colonnes configurées.

Pour appliquer des filtres aux colonnes, modifiez la propriété Filtre de colonne dans le contexte Paramètres de réplication, en ajoutant un tableau de configurations, une entrée pour chaque table à laquelle vous souhaitez appliquer un filtre.

Les colonnes peuvent être incluses ou exclues par nom ou par modèle. Vous pouvez appliquer une seule condition par table ou combiner plusieurs conditions, les exclusions ayant toujours la priorité sur les inclusions.

L’exemple suivant montre les champs disponibles. schema et table sont obligatoires, puis un ou plusieurs des champs included, excluded, includedPattern, excludedPattern sont requis.

[  {  "schema": "<source table schema>",  "table" : "<source table name>",  "included": ["<column name>", "<column name>"],  "excluded": ["<column name>", "<column name>"],  "includedPattern": "<regular expression>",  "excludedPattern": "<regular expression>",  } ] 

Suivre les changements de données dans les tables¶

Le connecteur réplique non seulement l’état actuel des données des tables sources, mais aussi l’état de chaque ligne de chaque jeu de modifications. Ces données sont stockées dans des tables de journal créées dans le même schéma que la table de destination.

Les noms des tables de journal sont formatés comme suit : <source table name>_JOURNAL_<timestamp>_<schema generation> where <timestamp> is the value of epoch seconds when the source table was added to replication, and <schema generation> est un nombre entier qui augmente à chaque modification du schéma sur la table source. Par conséquent, les tables sources qui subissent des modifications de schéma auront plusieurs tables de journal.

Lorsqu’une table est retirée de la réplication, puis réintroduite, la valeur de l’horodatage <> change et la génération du schéma <> reprend à partir de 1.

Le connecteur ne supprime jamais les tables de journal, mais il utilise le journal le plus récent pour chaque table source répliquée, en se contentant de lire les flux d’ajout uniquement au-dessus des journaux. Pour récupérer le stockage, vous pouvez :

• Tronquer toutes les tables de journal à tout moment.

• Supprimer les tables de journal liées aux tables sources qui ont été retirées de la réplication.

• Supprimer toutes les tables de journal de la dernière génération, sauf les tables activement répliquées.

Par exemple, si votre connecteur est paramétré pour répliquer activement la table source orders, et que vous avez précédemment supprimé la table customers de la réplication, vous pouvez avoir les tables de journal suivantes. Dans ce cas, vous pouvez toutes les supprimer, à l’exception de orders_5678_2.

customers_1234_1 customers_1234_2 orders_5678_1 orders_5678_2 

Configuration de la planification des tâches de fusion¶

Le connecteur utilise un entrepôt pour fusionner les données de capture des données de changement (CDC) dans les tables de destination. Cette opération est déclenchée par le processeur MergeSnowflakeJournalTable. S’il n’y a pas de nouvelles modifications ou si aucun nouveau fichier de flux n’est en attente dans la file d’attente MergeSnowflakeJournalTable, aucune fusion n’est déclenchée et l’entrepôt se suspend automatiquement.

Pour limiter le coût de l’entrepôt et restreindre les fusions aux seules heures planifiées, utilisez l’expression CRON dans le paramètre de planification CRON de la tâche de fusion. Elle limite les fichiers de flux arrivant au processeur MergeSnowflakeJournalTable et les fusions ne sont déclenchées qu’au cours d’une période donnée. Pour plus d’informations sur la planification, voir Stratégie de planification.

Arrêter ou supprimer le connecteur¶

Lors de l’arrêt ou de la suppression du connecteur, vous devez tenir compte de l’emplacement de réplication https://www.postgresql.org/docs/current/warm-standby.html#STREAMING-REPLICATION-SLOTS utilisé par le connecteur.

Le connecteur crée son propre slot de réplication avec un nom commençant par snowflake_connector_ suivi d’un suffixe aléatoire. À mesure que le connecteur lit le flux de réplication, il avance l’emplacement, de sorte que PostgreSQL puisse découper son journal WAL et libérer de l’espace disque.

Lorsque le connecteur est en pause, le slot n’est pas avancé et les modifications apportées à la base de données source continuent d’augmenter la taille du journal WAL. Vous ne devez pas laisser le connecteur en pause pendant de longues périodes, en particulier dans les bases de données à fort trafic.

Lorsque le connecteur est supprimé, que ce soit en le supprimant du canevas Openflow ou par tout autre moyen, comme la suppression de l’instance Openflow entière, le slot de réplication reste en place et doit être supprimé manuellement.

Si vous avez plusieurs instances de connecteur qui répliquent à partir de la même base de données PostgreSQL, chaque instance créera son propre slot de réplication portant un nom unique. Lorsque vous supprimez manuellement un emplacement de réplication, assurez-vous qu’il s’agit du bon. Vous pouvez voir quel emplacement de réplication est utilisé par une instance de connecteur donnée en vérifiant l’état du processeur CaptureChangePostgreSQL.