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 disposant du privilège CREATE dans la base de données et exécutez la requête suivante :

CREATE PUBLICATION <publication name>;

Copy

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

Pour plus d’informations, voir ALTER PUBLICATION.

5. En tant qu’administrateur de compte Snowflake, effectuez les tâches suivantes :

   1. Créez un utilisateur Snowflake avec le type SERVICE. Créez une base de données pour stocker les données répliquées et définissez les privilèges permettant à l’utilisateur de Snowflake de créer des objets dans cette base de données en lui accordant les privilèges USAGE et CREATE SCHEMA.

      CREATE DATABASE <destination_database>;
      CREATE USER <openflow_user> TYPE=SERVICE COMMENT='Service user for automated access of Openflow';
      CREATE ROLE <openflow_role>;
      GRANT ROLE <openflow_role> TO USER <openflow_user>;
      GRANT USAGE ON DATABASE <destination_database> TO ROLE <openflow_role>;
      GRANT CREATE SCHEMA ON DATABASE <destination_database> TO ROLE <openflow_role>;
      CREATE WAREHOUSE <openflow_warehouse>
           WITH
                WAREHOUSE_SIZE = 'MEDIUM'
                AUTO_SUSPEND = 300
                AUTO_RESUME = TRUE;
      GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
      

      Copy

   2. Créez une paire de clés sécurisées (publique et privée). Stockez la clé privée de l’utilisateur dans un fichier à fournir à la configuration du connecteur. Attribuez la clé publique à l’utilisateur du service Snowflake :

      ALTER USER <openflow_user> SET RSA_PUBLIC_KEY = 'thekey';
      

      Copy

      Pour plus d’informations, voir paire de clés.

   3. Désignez un entrepôt à utiliser par le connecteur. Commencez par la taille d’entrepôt MEDIUM, puis faites des essais en fonction du nombre de tables répliquées et de la quantité de données transférées. Les tables de grande taille s’adaptent généralement mieux aux entrepôts multi-clusters, plutôt qu’à la taille de l’entrepôt.

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.

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 du journal sont formatés comme suit : <nom de la table source>_JOURNAL_<horodatage>_<génération du schéma>

où <horodatage> est la valeur en secondes de l’époque à laquelle la table source a été ajoutée à la réplication, et <génération de schéma> est un nombre entier augmentant à chaque changement de schéma sur la table source. Cela signifie qu’une table source qui subit des modifications de schéma aura 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 n’abandonne jamais les tables de logs, mais il n’utilise activement que le dernier journal pour chaque table source répliquée. Si vous souhaitez récupérer le stockage, vous pouvez en toute sécurité supprimer les tables de journal liées aux tables sources qui ont été retirées de la réplication, ainsi que toutes les tables activement répliquées, à l’exception de celles de la dernière génération.

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.