Paramétrez Openflow Connector for MySQL¶

Conditions préalables¶

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

2. Assurez-vous que vous disposez de MySQL 8 ou d’une version ultérieure pour synchroniser les données avec Snowflake.

3. Assurez-vous que vous avez paramétré Openflow.

4. En tant qu’administrateur de la base de données, effectuez les tâches suivantes :

   1. Activez les logs binaires, puis enregistrez et configurez leur format comme suit :

      log_bin	Définissez sur : on. Cela permet d’activer le journal binaire qui enregistre les modifications structurelles et les modifications apportées aux données.
      binlog_format	Définissez sur : row. Le connecteur ne prend en charge que la réplication basée sur les lignes. Il se peut que les versions 8.x de MySQL soient les dernières à prendre en charge ce paramètre, et les versions futures ne prendront en charge que la réplication basée sur les lignes. Ne s’applique pas dans GCP Cloud SQL, où il est défini sur la bonne valeur.
      binlog_row_metadata	Définissez sur : full. Le connecteur a besoin de toutes les métadonnées des lignes pour fonctionner, en particulier des noms des colonnes et des informations sur les clés primaires.
      binlog_row_image	Définissez sur : full. Le connecteur nécessite que toutes les colonnes soient écrites dans le journal binaire. Ne s’applique pas dans Amazon Aurora, où cela est défini sur la bonne valeur.
      binlog_row_value_options	Laissez vide. Cette option n’affecte que les colonnes JSON, où elle peut être définie de sorte à n’inclure que les parties modifiées des documents JSON pour les instructions UPDATE. Le connecteur nécessite que les documents complets soient écrits dans le journal binaire.
      binlog_expire_logs_seconds	Définissez cette option au minimum sur quelques heures ou sur une durée supérieure pour garantir que l’agent de base de données puisse poursuivre la réplication incrémentielle après des pauses prolongées ou des temps d’arrêt. Snowflake vous recommande de fixer la période d’expiration des logs binaires (binlog_expire_logs_seconds) à au moins quelques heures afin de garantir un fonctionnement stable du connecteur. À l’issue de la période d’expiration du journal binaire, les fichiers du journal binaire peuvent être automatiquement supprimés. Si l’intégration est interrompue pendant une longue période, par exemple en raison de travaux de maintenance, et que les journaux binaires expirés sont supprimés pendant cette période, Openflow ne sera pas en mesure de répliquer les données de ces fichiers. Si vous utilisez la réplication planifiée, la valeur doit être supérieure à la planification configurée.

      Voyez le code suivant à titre d’exemple :

      log_bin = on
      binlog_format = row
      binlog_row_metadata = full
      binlog_row_image = full
      binlog_row_value_options =
      

      Copy

   2. Augmentez la valeur de sort_buffer_size.

      sort_buffer_size = 4194304
      

      Copy

      sort_buffer_size définit la quantité de mémoire (en octets) allouée par fil de requête pour les opérations de tri en mémoire, telles que ORDER BY. Si la valeur est trop petite, le connecteur peut échouer avec le message d’erreur suivant :

      Out of sort memory, consider increasing server sort buffer size. Cela indique que sort_buffer_size devrait être augmenté.

   3. Si vous utilisez les bases de données Amazon RDS, augmentez la période de conservation pertinente pour binlog_expire_logs_seconds en utilisant rds_set_configuration. Par exemple, si vous souhaitez stocker le binlog pendant 24 heures, appelez mysql.rds_set_configuration('binlog retention hours', 24).

   4. Connexion via SSL. Si vous planifiez d’utiliser une connexion SSL vers MySQL, préparez le certificat racine de votre serveur de base de données. Elle est exigée lors de la configuration.

   5. Créez un utilisateur pour le connecteur. Le connecteur nécessite un utilisateur disposant des privilèges REPLICATION_SLAVE et REPLICATION_CLIENT pour la lecture des logs binaires. Accordez ces privilèges :

      GRANT REPLICATION SLAVE ON *.* TO '<username>'@'%'
      GRANT REPLICATION CLIENT ON *.* TO '<username>'@'%'
      

      Copy

   6. Accordez le privilège SELECT sur chaque table répliquée :

      GRANT SELECT ON <schema>.* TO '<username>'@'%'
      GRANT SELECT ON <schema>.<table> TO '<username>'@'%'
      

      Copy

      Pour plus d’informations sur la sécurité de la réplication, voir journal binaire.

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.

Définir le connecteur¶

En tant qu’ingénieur des données, effectuez les tâches suivantes pour installer et configurer le connecteur :

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

customers_1234_1
customers_1234_2
orders_5678_1
orders_5678_2

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.