Paramétrez Openflow Connector for SQL Server¶

Conditions préalables¶

Avant de configurer le connecteur, assurez-vous d’avoir rempli les conditions préalables suivantes :

1. Assurez-vous d’avoir consulté À propos de Openflow Connector for SQL Server.

2. Assurez-vous d’avoir consulté Versions du serveur SQL prises en charge.

3. Assurez-vous d’avoir configuré votre déploiement d’exécution. Pour plus d’informations, consultez les rubriques suivantes :

   • Configuration d’Openflow - BYOC

   • Configurer Openflow - Déploiements Snowflake.

4. Si vous utilisez Openflow - Snowflake Deployments, assurez-vous d’avoir examiné la configuration des domaines requis et d’avoir accordé l’accès aux domaines requis pour le connecteur SQL Server.

Configurer votre instance SQL Server¶

Avant de configurer le connecteur, effectuez les tâches suivantes dans votre environnement SQL Server :

1. Activez le suivi des modifications sur les bases de données et les tables que vous prévoyez de répliquer, comme montré dans l’exemple SQL Server suivant :

   CopyExpand

   ALTER DATABASE <database>
     SET CHANGE_TRACKING = ON
     (CHANGE_RETENTION = 2 DAYS, AUTO_CLEANUP = ON);
   
   ALTER TABLE <schema>.<table>
     ENABLE CHANGE_TRACKING;
   

   Show lessSee more

   Scroll to top

   Note

   Exécutez ces commandes pour chaque base de données et chaque table que vous prévoyez de répliquer.

   Le connecteur exige que le suivi des modifications soit activé sur les bases de données et les tables avant le début de la réplication. Assurez-vous que le suivi des modifications est activé pour chaque table que vous prévoyez de répliquer. Vous pouvez également activer le suivi des modifications pour d’autres tables lorsque le connecteur est en cours d’exécution.

2. Créez une connexion pour l’instance SQL Server :

   CopyExpand

   CREATE LOGIN <user_name> WITH PASSWORD = '<password>';
   

   Show lessSee more

   Scroll to top

   Cette connexion est utilisée pour créer des utilisateurs pour les bases de données que vous prévoyez de répliquer.

3. Créez un utilisateur pour chaque base de données que vous répliquez en exécutant la commande SQL Server suivante dans chaque base de données :

   CopyExpand

   USE <source_database>;
   CREATE USER <user_name> FOR LOGIN <user_name>;
   

   Show lessSee more

   Scroll to top

4. Accordez les autorisations SELECT et VIEW CHANGE TRACKING à l’utilisateur pour chaque base de données que vous répliquez :

   CopyExpand

   GRANT SELECT ON <database>.<schema>.<table> TO <user_name>;
   GRANT VIEW CHANGE TRACKING ON <database>.<schema>.<table> TO <user_name>;
   

   Show lessSee more

   Scroll to top

   Exécutez ces commandes dans chaque base de données pour chaque table que vous prévoyez de répliquer. Ces autorisations doivent être accordées à l’utilisateur de chaque base de données que vous avez créée précédemment.

5. (Facultatif) Accordez le privilège VIEW DEFINITION sur les types de données définis par l’utilisateur (UDDT).

   Si vos tables contiennent des colonnes qui utilisent des types de données définis par l’utilisateur (UDDT), et si ces UDDT appartiennent à un autre utilisateur que l’utilisateur du connecteur, vous devez accorder l’autorisation VIEW DEFINITION à l’utilisateur du connecteur, comme indiqué dans l’exemple suivant SQL Server :

   CopyExpand

   GRANT VIEW DEFINITION TO <user_name>;
   

   Show lessSee more

   Scroll to top

   Sans cette autorisation, les colonnes utilisant des UDDT sont exclues silencieusement de la réplication.

6. (Facultatif) Configurez la connexion SSL.

   Si vous utilisez une connexion SSL pour vous connecter au SQL Server, créez le certificat racine pour votre serveur de base de données. Ceci est nécessaire lors de la configuration du connecteur.

Configurer votre environnement Snowflake¶

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

1. Créez une base de données de destination dans Snowflake pour stocker les données répliquées :

   CopyExpand

   CREATE DATABASE <destination_database>;
   

   Show lessSee more

   Scroll to top

2. Créez un utilisateur de service Snowflake :

   CopyExpand

   CREATE USER <openflow_user>
     TYPE = SERVICE
     COMMENT='Service user for automated access of Openflow';
   

   Show lessSee more

   Scroll to top

3. Créez un rôle Snowflake pour le connecteur et accordez-lui les privilèges requis :

   CopyExpand

   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>;
   

   Show lessSee more

   Scroll to top

   Utilisez ce rôle pour gérer l’accès du connecteur à la base de données Snowflake.

   Pour créer des objets dans la base de données de destination, vous devez accorder les privilèges USAGE et CREATE SCHEMA sur la base de données au rôle utilisé pour gérer l’accès.

4. Créez un entrepôt Snowflake pour le connecteur et accordez-lui les privilèges requis :

   CopyExpand

   CREATE WAREHOUSE <openflow_warehouse> WITH
     WAREHOUSE_SIZE = 'XSMALL'
     AUTO_SUSPEND = 300
     AUTO_RESUME = TRUE;
   GRANT USAGE, OPERATE ON WAREHOUSE <openflow_warehouse> TO ROLE <openflow_role>;
   

   Show lessSee more

   Scroll to top

   Snowflake recommande de commencer par une taille d’entrepôt XSMALL, puis de tester différentes tailles en fonction du nombre de tables répliquées et de la quantité de données transférées. Un grand nombre de tables s’adaptent généralement mieux aux entrepôts multi-clusters qu’à un entrepôt de plus grande taille. Pour plus d’informations, consultez la section relative aux entrepôts multi-clusters.

5. Configurez les clés publiques et privées pour l’authentification par paire de clés :

   1. Créez une paire de clés sécurisées (publique et privée).

   2. Stockez la clé privée de l’utilisateur dans un fichier à fournir à la configuration du connecteur.

   3. Attribuez la clé publique à l’utilisateur du service Snowflake :

      CopyExpand

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

      Show lessSee more

      Scroll to top

      Pour plus d’informations, voir Authentification par paire de clés et rotation de paires de clés.

Installer le connecteur¶

Pour installer le connecteur, procédez comme suit en tant qu’ingénieur des données :

1. Accédez à la page d’aperçu d’Openflow. Dans la section Featured connectors, sélectionnez View more connectors.

2. Sur la page des connecteurs Openflow, trouvez le connecteur et sélectionnez Add to runtime.

3. Dans la boîte de dialogue Select runtime, sélectionnez votre environnement d’exécution dans la liste déroulante Available runtimes, puis cliquez sur Add.

   Note

   Avant d’installer le connecteur, assurez-vous que vous avez créé une base de données et un schéma dans Snowflake pour que le connecteur puisse stocker les données ingérées.

4. Authentifiez-vous au déploiement avec les identifiants de votre compte Snowflake et sélectionnez Allow lorsque vous êtes invité à autoriser l’application d’exécution à accéder à votre compte Snowflake. Le processus d’installation du connecteur prend quelques minutes.

5. Authentifiez-vous auprès de l’environnement d’exécution avec les identifiants de votre compte Snowflake.

Le canevas Openflow apparaît avec le groupe de processus du connecteur ajouté.

Configuration du connecteur¶

Pour configurer le connecteur, procédez comme suit en tant qu’ingénieur des données :

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

2. Remplissez les valeurs de paramètres requises.

   Pour plus d’informations sur les valeurs de paramètres requises, reportez-vous aux sections suivantes :

   • Paramètres source SQLServer : Utilisé pour établir une connexion avec SQL Server.

   • Paramètres de destination SQLServer : Utilisé pour établir une connexion avec Snowflake.

   • Paramètres d’ingestion SQLServer : Pour spécifier les tables à répliquer.

Commencez par définir les paramètres du contexte des paramètres source SQLServer, puis définissez le contexte des paramètres de destination SQLServer. Une fois cette opération terminée, activez le connecteur. Le connecteur se connecte à la fois à SQLServer et à Snowflake et démarre l’exécution. Toutefois, le connecteur ne réplique aucune donnée tant que les tables à répliquer n’ont pas été explicitement ajoutées à sa configuration.

Pour configurer des tables spécifiques pour la réplication, modifiez le contexte des paramètres d’ingestion SQLServer. Une fois que vous avez appliqué les modifications au contexte des paramètres d’ingestion SQLServer, la configuration est reprise par le connecteur et le cycle de vie de la réplication démarre pour chaque table.

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

Le connecteur peut ingérer des données d’un serveur principal ou d’un serveur d’abonnement à l’aide d’une réplication transactionnelle. Avant de configurer le connecteur pour qu’il se connecte à une réplique SQL Server, assurez-vous que la réplication entre les nœuds principaux et les nœuds de réplique fonctionne correctement. Pour obtenir des instructions sur la configuration de la réplication transactionnelle, consultez Tutoriel : Configurer la réplication transactionnelle. 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 et les événements de suivi des modifications sont présents dans le serveur de réplique utilisé par le connecteur.

Pour configurer le connecteur afin qu’il lise à partir d’un serveur d’abonnés au lieu d’un serveur de l’éditeur, spécifiez l’URL du serveur d’abonnés dans le paramètre SQLServer Connection URL.

Redémarrer la réplication de table¶

Une table à l’état FAILED, par exemple, en raison d’une clé primaire manquante ou d’un changement de schéma non pris en charge, ne redémarre pas automatiquement. Si une table passe à l’état FAILED ou si vous devez redémarrer la réplication à partir de zéro, utilisez la procédure suivante pour supprimer et ajouter à nouveau la table à la réplication.

1. Supprimez la table des paramètres de flux. Dans le contexte Paramètres d’ingestion, supprimez la table dans les Included Table Names ou modifiez le Included Table Regex de sorte que la table n’a plus aucune correspondance.

2. Vérifiez que la table a été supprimée :

   1. Dans le canevas de l’environnement d’exécution Openflow, cliquez avec le bouton droit de la souris sur un groupe de processeurs et choisissez Controller Services.

   2. Dans la table répertoriant les services du contrôleur, recherchez la ligne Table State Store, cliquez sur les trois points verticaux sur le côté droit de la ligne, puis choisissez View State.

   Important

   Vous devez attendre que l’état de la table soit entièrement supprimé de cette liste avant de poursuivre. Ne continuez pas tant que cette modification de la configuration n’est pas terminée.

3. Nettoyez la destination. Une fois que l’état de la table indique qu’elle est entièrement supprimés, DROP manuellement la table de destination dans Snowflake. Notez que le connecteur ne remplacera pas une table de destination existante pendant la phase de l’instantané. Si la table existe toujours, la réplication échouera à nouveau. En option, la table et le flux de journal peuvent également être supprimés s’ils ne sont plus nécessaires.

4. Rajouter la table : Mettez à jour les paramètres Included Table Names ou Included Table Regex pour inclure à nouveau la table.

5. Vérifiez le redémarrage. Vérifiez le Table State Store en utilisant les instructions données précédemment. L’état de la table doit apparaître avec le statut NEW, puis SNAPSHOT_REPLICATION et pour finir INCREMENTAL_REPLICATION.

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

Le connecteur peut filtrer les données répliquées par table vers un sous-ensemble de colonnes configurées. Les colonnes de clé primaire sont toujours incluses, quelles que soient les exclusions.

Pour appliquer des filtres de colonnes, définissez le paramètre Column Filter JSON dans le contexte Paramètres d’ingestion sur un tableau JSON d’objets de filtre, à raison d’un par table que vous souhaitez filtrer.

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

[
    {
        "database": "<database>" | "databasePattern": "<regex>",
        "schema": "<schema>" | "schemaPattern": "<regex>",
        "table": "<table>" | "tablePattern": "<regex>",
        "included": ["<column>", "<column>"],
        "excluded": ["<column>", "<column>"],
        "includedPattern": "<regex>",
        "excludedPattern": "<regex>"
    }
]

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "orders",
        "included": ["account_id", "status", "created_at"]
    }
]

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "orders",
        "excluded": ["internal_note", "debug_flag"]
    }
]

[
    {
        "database": "my_db",
        "schema": "dbo",
        "table": "contacts",
        "includedPattern": ".*_email",
        "excluded": ["admin_email"]
    }
]

[
    {
        "databasePattern": "prod_.*",
        "schema": "dbo",
        "table": "customers",
        "excluded": ["internal_note"]
    }
]

[
    {"database": "my_db", "schema": "dbo", "table": "orders", "included": ["account_id", "status"]},
    {"database": "my_db", "schema": "dbo", "table": "customers", "excludedPattern": ".*_internal"}
]

Réplication d’une table partitionnée¶

Le connecteur prend en charge la réplication des tables partitionnées. Une table partitionnée du serveur SQL est répliquée dans Snowflake sous la forme d’une seule table de destination, contenant les données de toutes les partitions.

Pour répliquer une table partitionnée, assurez-vous que le suivi des modifications est activé sur la table partitionnée, comme décrit dans Configurer votre instance SQL Server.

Suivre les changements de données dans les tables¶

Le connecteur réplique l’état actuel des données des tables sources, ainsi que les modifications détectées à partir de chaque intervalle d’interrogation. 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> où <timestamp> correspond au nombre de secondes écoulées depuis l’époque de référence au moment où la table source a été ajoutée à la réplication, et <schema_generation> est un nombre entier qui augmente à chaque modification du schéma de la table source. Par conséquent, les tables sources qui subissent des modifications de schéma auront plusieurs tables de journal.

Lorsque vous supprimez une table de la réplication, puis que vous l’ajoutez à nouveau, la valeur <timestamp> change, et <schema_generation> recommence à 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.

Utilisez l’expression CRON dans le paramètre de planification CRON de la tâche de fusion pour limiter le coût de l’entrepôt et restreindre les fusions aux seules heures planifiées. 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.

Exécutez le flux¶

1. Cliquez avec le bouton droit de la souris sur l’avion et sélectionnez Enable all Controller Services.

2. Cliquez avec le bouton droit de la souris sur le groupe de processus importé et sélectionnez Start. Le connecteur démarre l’ingestion des données.