Réplication des zones de préparation, des canaux et de l’historique des chargements¶

Exigences¶

Avant de pouvoir répliquer les objets du pipeline de données, vous devez activer la réplication ETL. La prise en charge de la réplication ETL fait partie du bundle 2024_01 BCR. Vous pouvez activer le bundle 2024_01 BCR en exécutant l’instruction suivante :

SELECT SYSTEM$ENABLE_BEHAVIOR_CHANGE_BUNDLE('2024_01');

Pour répliquer toutes les zones de préparation externes qui utilisent une intégration de stockage, vous devez également configurer votre groupe de réplication ou de basculement pour répliquer STORAGE INTEGRATIONS. Dans le cas contraire, les zones de préparation externes sont répliquées sans l’intégration de stockage associée.

Vous pouvez utiliser une instruction ALTER REPLICATION GROUP ou ALTER FAILOVER GROUP pour modifier ces propriétés pour un groupe existant.

Par exemple :

ALTER FAILOVER GROUP my_failover_group SET
  OBJECT_TYPES = ROLES, INTEGRATIONS
  ALLOWED_INTEGRATION_TYPES = API INTEGRATIONS, STORAGE INTEGRATIONS;

Réplication et zones de préparation¶

Cette section décrit le niveau actuel de fonctionnalité de réplication que Snowflake prend en charge pour différents types de zones de préparation.

Réplication et canaux¶

Cette section décrit le niveau actuel des fonctionnalités de réplication prises en charge pour les différents types de canaux.

Snowflake prend en charge la réplication pour les éléments suivants :

• Objets « canaux », y compris les canaux « intégration automatique » et des points de terminaison REST qui chargent des données à partir de zones de préparation externes.

• Paramètres au niveau du canal.

• Octroi de privilèges sur les objets de type canal.

Exemple 1 : Répliquer une zone de préparation interne nommée¶

Cet exemple montre comment fonctionne la réplication pour les zones de préparation internes. En particulier, l’exemple montre comment la table de répertoire est la seule source de vérité pour les métadonnées de table avant et après la réplication.

La première partie de l’exemple effectue les tâches suivantes dans un compte source.

1. Créer une zone de préparation interne nommée my_int_stage avec une table de répertoire activée pour répliquer les fichiers sur la zone de préparation. Copier ensuite les données d’une table nommée my_table dans des fichiers sur la zone de préparation.

   Note

   L’exemple actualise la table de répertoire après avoir chargé file1 et file2 sur la zone de préparation afin de synchroniser les métadonnées de la table avec le dernier ensemble de fichiers dans la définition de la zone de préparation pour les tables de répertoire. Cependant, aucune opération d’actualisation ne se produit après le chargement de file3.

   CREATE OR REPLACE STAGE my_stage
     DIRECTORY = (ENABLE = TRUE);
   
   COPY INTO @my_stage/folder1/file1 from my_table;
   COPY INTO @my_stage/folder2/file2 from my_table;
   ALTER STAGE my_stage REFRESH;
   
   COPY INTO @my_stage/folder3/file3 from my_table;
   

   Copy

2. Créez un groupe de basculement :

   CREATE FAILOVER GROUP my_stage_failover_group
     OBJECT_TYPES = DATABASES
     ALLOWED_DATABASES = my_database_1
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

La deuxième partie de l’exemple complète le processus de réplication et de basculement dans un compte cible :

1. Créer un groupe de basculement en tant que réplique du groupe de basculement du compte source, actualiser les objets du nouveau groupe de basculement et faire en sorte que le compte cible serve de compte source.

   CREATE FAILOVER GROUP my_stage_failover_group
     AS REPLICA OF myorg.my_account_1.my_stage_failover_group;
   
   ALTER FAILOVER GROUP my_stage_failover_group REFRESH;
   
   ALTER FAILOVER GROUP my_stage_failover_group PRIMARY;
   

   Copy

2. Ensuite, actualiser la table de répertoire sur la zone de préparation répliquée et copier tous les fichiers suivis par la table de répertoire sur my_stage dans une table nommée my_table .

   Note

   L’instruction COPY INTO charge file1 et file2 dans la table, mais pas file3. Cela est dû au fait que la table de répertoire n’a pas été actualisée après l’ajout de file3 dans le compte source.

   ALTER STAGE my_stage REFRESH;
   
   COPY INTO my_table FROM @my_stage;
   

   Copy

Exemple 2 : répliquer une zone de préparation externe et une intégration de stockage¶

Cet exemple fournit un exemple de flux de travail pour la réplication d’une zone de préparation externe et d’une intégration de stockage vers un compte cible.

L’exemple suppose que vous avez déjà effectué les opérations suivantes : Configurer l’accès sécurisé à votre compartiment Amazon S3.

La première partie de l’exemple effectue les tâches suivantes dans un compte source.

1. Créer une intégration de stockage pour un compartiment Amazon S3 dans la base de données my_database_2.

   CREATE STORAGE INTEGRATION my_storage_int
     TYPE = external_stage
     STORAGE_PROVIDER = 's3'
     STORAGE_ALLOWED_LOCATIONS = ('s3://mybucket/path')
     STORAGE_BLOCKED_LOCATIONS = ('s3://mybucket/blockedpath')
     ENABLED = true;
   

   Copy

2. Créer une zone de préparation externe dans la base de données my_database_2 en utilisant l’intégration de stockage my_storage_int.

   CREATE STAGE my_ext_stage
     URL = 's3://mybucket/path'
     STORAGE_INTEGRATION = my_storage_int
   

   Copy

3. Créer un groupe de basculement et inclure la base de données my_database_2 et les objets d’intégration de stockage.

   CREATE FAILOVER GROUP my_external_stage_fg
     OBJECT_TYPES = databases, integrations
     ALLOWED_INTEGRATION_TYPES = storage integrations
     ALLOWED_DATABASES = my_database_2
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

La deuxième partie de l’exemple complète le processus de réplication et de basculement dans un compte cible :

1. Créer un groupe de basculement comme une réplique du groupe de basculement dans le compte source et actualiser.

   CREATE FAILOVER GROUP my_external_stage_fg
     AS REPLICA OF myorg.my_account_1.my_external_stage_fg;
   
   ALTER FAILOVER GROUP my_external_stage_fg REFRESH;
   

   Copy

2. Après avoir répliqué l’intégration de stockage sur le compte cible, vous devez prendre des mesures supplémentaires pour mettre à jour les autorisations de votre fournisseur Cloud afin d’accorder à l’intégration de réplication l’accès à votre stockage sur le Cloud. Pour plus d’informations, voir Configurer l’accès au stockage Cloud pour les intégrations de stockage secondaire.

Exemple 3 : répliquer un canal d’intégration automatique¶

Cet exemple fournit un exemple de workflow pour répliquer un canal qui utilise un sujet Amazon Simple Notification Service (SNS) avec Amazon Simple Queue Service (SQS) pour automatiser Snowpipe.

L’exemple suppose que vous avez déjà effectué les tâches suivantes :

• Création et configuration d’une intégration de stockage pour Amazon S3. À titre d’exemple, nous utilisons une intégration de stockage nommée my_s3_storage_int.

• Création d’un sujet et d’un abonnement Amazon SNS, et abonnement de la file d’attente Snowflake SQS à votre sujet SNS.

• Création d’une zone de préparation externe qui fait référence à votre intégration de stockage. À titre d’exemple, nous utilisons une zone de préparation nommée my_s3_stage. Pour obtenir des instructions, voir CREATE STAGE.

Commencer par les tâches suivantes dans un compte source.

1. Utiliser la commande CREATE PIPE pour créer un canal avec l’intégration automatique activée qui charge les données de la zone de préparation externe dans une table nommée mytable.

   CREATE PIPE snowpipe_db.public.mypipe AUTO_INGEST=TRUE
    AWS_SNS_TOPIC='<topic_arn>'
    AS
      COPY INTO snowpipe_db.public.mytable
      FROM @snowpipe_db.public.my_s3_stage
      FILE_FORMAT = (TYPE = 'JSON');
   

   Copy

2. Actualiser le canal avec une instruction ALTER PIPE pour charger les données de la zone de préparation des 7 derniers jours.

   ALTER PIPE mypipe REFRESH;
   

   Copy

3. Enfin, utiliser CREATE FAILOVER GROUP pour créer un groupe de basculement qui permet la réplication des intégrations de stockage.

   CREATE FAILOVER GROUP my_pipe_failover_group
     OBJECT_TYPES = DATABASES, INTEGRATIONS
     ALLOWED_INTEGRATION_TYPES = STORAGE INTEGRATIONS
     ALLOWED_DATABASES = snowpipe_db
     ALLOWED_ACCOUNTS = myorg.my_account_2;
   

   Copy

La deuxième partie de l’exemple complète le processus de réplication et de basculement dans un compte cible :

1. Créer un groupe de basculement comme une réplique du groupe de basculement dans le compte source.

   CREATE FAILOVER GROUP my_pipe_failover_group
     AS REPLICA OF myorg.my_account_1.my_pipe_failover_group;
   

   Copy

2. Exécuter une instruction DESCRIBE INTEGRATION pour récupérer l’ARN de l’utilisateur AWS IAM pour votre compte Snowflake sur le déploiement secondaire.

   Utiliser l’ARN pour accorder à l’utilisateur les autorisations IAMpour accéder à votre compartiment S3. Voir l’étape 5 : Accorder à l’utilisateur les autorisations IAM pour accéder aux objets du compartiment.

   DESC INTEGRATION my_s3_storage_int;
   

   Copy

3. Appeler la SYSTEM$GET_AWS_SNS_IAM_POLICY fonction système pour générer une politique IAM qui accorde à la nouvelle file d’attente SQS l’autorisation de s’abonner à votre sujet SNS. Snowflake a créé la nouvelle file d’attente SQS dans votre compte cible lorsque vous avez répliqué le groupe de basculement à partir de votre compte source.

   SELECT SYSTEM$GET_AWS_SNS_IAM_POLICY('<topic_arn>');
   

   Copy

   topic_arn est le nom de ressource Amazon (ARN) du sujet SNS que vous avez créé pour le canal d’origine dans votre compte source.

   Ensuite, souscrire la nouvelle file d’attente Amazon SQS à votre sujet SNS.

4. Actualiser les objets de votre nouveau groupe de basculement.

   ALTER FAILOVER GROUP my_pipe_failover_group REFRESH;
   

   Copy

5. Enfin, la commande ALTER FAILOVER GROUP vous permet de promouvoir le compte cible en tant que compte source.

   ALTER FAILOVER GROUP my_pipe_failover_group PRIMARY;
   

   Copy

   Le canal mypipe commencera à charger toutes les données qui ont été rendues disponibles depuis la dernière fois que le groupe de basculement a été actualisé dans le compte source.

   Pour vérifier que le canal répliqué fonctionne, interrogez la table à partir de l’instruction COPY du canal.

   SELECT * FROM mytable;
   

   Copy