Snowpipe automatisé pour Google Cloud Storage¶

Étape 1 : Création d’une intégration Cloud Storage dans Snowflake¶

Créez une intégration à l’aide de la commande CREATE STORAGE INTEGRATION. Une intégration est un objet Snowflake qui délègue la responsabilité de l’authentification pour un stockage externe dans le Cloud à une entité générée par Snowflake (c’est-à-dire un compte de service Cloud Storage). Pour accéder aux compartiments Cloud Storage, Snowflake crée un compte de service auquel des autorisations peuvent être accordées pour accéder aux compartiments dans lesquels vos fichiers de données sont stockés.

Une seule intégration de stockage peut prendre en charge plusieurs zones de préparation externes (c.-à-d. GCS). L’URL dans la définition de zone de préparation doit correspondre aux compartiments GCS (et aux chemins facultatifs) spécifiés pour le paramètre STORAGE_ALLOWED_LOCATIONS.

CREATE STORAGE INTEGRATION <integration_name>
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'GCS'
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('gcs://<bucket>/<path>/', 'gcs://<bucket>/<path>/')
  [ STORAGE_BLOCKED_LOCATIONS = ('gcs://<bucket>/<path>/', 'gcs://<bucket>/<path>/') ]

Où :

• integration_name est le nom de la nouvelle intégration.

• bucket est le nom d’un compartiment Cloud Storage qui stocke vos fichiers de données (par exemple, mybucket). Les paramètres STORAGE_ALLOWED_LOCATIONS requis et STORAGE_BLOCKED_LOCATIONS facultatif limitent ou bloquent l’accès à ces compartiments, respectivement, lors de la création ou de la modification de zones de préparation faisant référence à cette intégration.

• path est un chemin facultatif qui peut être utilisé pour fournir un contrôle granulaire sur les objets du compartiment.

L’exemple suivant crée une intégration qui limite explicitement les zones de préparation externes utilisant l’intégration pour faire référence à l’un des deux compartiments et des chemins. Dans une étape ultérieure, nous allons créer une zone de préparation externe qui fait référence à l’un de ces compartiments et chemins.

Les zones de préparation externes supplémentaires qui utilisent également cette intégration peuvent faire référence aux compartiments et aux chemins autorisés :

CREATE STORAGE INTEGRATION gcs_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'GCS'
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('gcs://mybucket1/path1/', 'gcs://mybucket2/path2/')
  STORAGE_BLOCKED_LOCATIONS = ('gcs://mybucket1/path1/sensitivedata/', 'gcs://mybucket2/path2/sensitivedata/');

Copy

Étape 2 : Récupération du compte de service Cloud Storage pour votre compte Snowflake¶

Exécutez la commande DESCRIBE INTEGRATION pour extraire l’ID du compte de service Cloud Storage créé automatiquement pour votre compte Snowflake :

DESC STORAGE INTEGRATION <integration_name>;

Où :

• integration_name est le nom de l’intégration créée à l”étape 1 : Création d’une intégration Cloud Storage dans Snowflake (ce chapitre).

Par exemple :

DESC STORAGE INTEGRATION gcs_int;

+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+
| property                    | property_type | property_value                                                              | property_default |
+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------|
| ENABLED                     | Boolean       | true                                                                        | false            |
| STORAGE_ALLOWED_LOCATIONS   | List          | gcs://mybucket1/path1/,gcs://mybucket2/path2/                               | []               |
| STORAGE_BLOCKED_LOCATIONS   | List          | gcs://mybucket1/path1/sensitivedata/,gcs://mybucket2/path2/sensitivedata/   | []               |
| STORAGE_GCP_SERVICE_ACCOUNT | String        | service-account-id@project1-123456.iam.gserviceaccount.com                  |                  |
+-----------------------------+---------------+-----------------------------------------------------------------------------+------------------+

Copy

La propriété STORAGE_GCP_SERVICE_ACCOUNT de la sortie affiche le compte de service Cloud Storage créé pour votre compte Snowflake (par exemple, service-account-id@project1-123456.iam.gserviceaccount.com). Nous fournissons un seul compte de service Cloud Storage pour l’ensemble de votre compte Snowflake. Toutes les intégrations Cloud Storage utilisent ce compte de service.

Étape 3 : Octroi d’autorisations au compte de service pour accéder à des objets de compartiment¶

Les instructions étape par étape suivantes décrivent comment configurer les permissions d’accès IAM pour Snowflake dans votre console Google Cloud Platform de sorte que vous puissiez utiliser un compartiment Cloud Storage pour charger et décharger les données :

Conditions préalables¶

Les instructions de cette rubrique supposent que les éléments suivants ont été créés et configurés :

Compte GCP:

• Sujet Pub/Sub qui reçoit des messages d’événement depuis le compartiment GCS. Pour plus d’informations, voir Création du sujet Pub/Sub (dans cette rubrique).

• Abonnement qui reçoit les messages d’événement du sujet Pub/Sub. Pour plus d’informations, voir Création de l’abonnement Pub/Sub (dans cette rubrique).

Pour obtenir des instructions, voir Documentation Pub/Sub.

Snowflake:

• Table cible dans la base de données Snowflake où vous souhaitez charger les données.

$ gsutil notification create -t <topic> -f json gs://<bucket-name> -e OBJECT_FINALIZE

Étape 1 : créer une intégration de notification dans Snowflake¶

Créez une intégration de notification à l’aide de la commande CREATE NOTIFICATION INTEGRATION. L’intégration des notifications fait référence à votre abonnement Pub/Sub. Snowflake associe l’intégration de la notification à un compte de service GCS créé pour votre compte. Snowflake crée un seul compte de service référencé par toutes les notifications de stockage GCS de votre compte Snowflake.

CREATE NOTIFICATION INTEGRATION <integration_name>
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = GCP_PUBSUB
  ENABLED = true
  GCP_PUBSUB_SUBSCRIPTION_NAME = '<subscription_id>';

Où :

• integration_name est le nom de la nouvelle intégration.

• subscription_id est le nom de l’abonnement que vous avez enregistré dans Récupération de l’ID d’abonnement Pub/Sub.

Par exemple :

CREATE NOTIFICATION INTEGRATION my_notification_int
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = GCP_PUBSUB
  ENABLED = true
  GCP_PUBSUB_SUBSCRIPTION_NAME = 'projects/project-1234/subscriptions/sub2';

Étape 2 : Accorder un accès à Snowflake à l’abonnement Pub/Sub¶

1. Exécutez la commande DESCRIBE INTEGRATION pour récupérer l’ID de compte de service Snowflake :

   DESC NOTIFICATION INTEGRATION <integration_name>;
   

   Copy

   Où :

   • integration_name est le nom de l’intégration créée à l”étape 1 : Créer une intégration de notification dans Snowflake.

   Par exemple :

   DESC NOTIFICATION INTEGRATION my_notification_int;
   

   Copy

2. Enregistrez le nom du compte de service dans la colonne GCP_PUBSUB_SERVICE_ACCOUNT, qui a le format suivant :

   <service_account>@<project_id>.iam.gserviceaccount.com
   

   Copy

3. Connectez-vous à la console Google Cloud Platform en tant qu’éditeur de projet.

4. Dans le tableau de bord d’accueil, sélectionnez Big Data » Pub/Sub » Subscriptions.

5. Sélectionnez l’abonnement à configurer pour l’accès.

6. Cliquez sur SHOW INFO PANEL dans le coin supérieur droit. Le panneau d’information de l’abonnement s’affiche en coulissant.

7. Cliquez sur le bouton ADD PRINCIPAL.

8. Dans le champ New principals, recherchez le nom du compte de service que vous avez enregistré.

9. Dans la liste déroulante Select a role, sélectionnez Pub/Sub Subscriber.

10. Cliquez sur le bouton Save. Le nom du compte de service est ajouté à la liste déroulante des rôles Pub/Sub Subscriber dans le panneau d’informations.

11. Accédez à la page Dashboard de la console Cloud et sélectionnez votre projet dans la liste déroulante.

12. Cliquez sur le bouton ADD PEOPLE TO THIS PROJECT .

13. Ajoutez le nom du compte de service que vous avez enregistré.

14. Dans la liste déroulante Select a role, sélectionnez Monitoring Viewer.

15. Cliquez sur le bouton Save . Le nom du compte de service est ajouté au rôle Monitoring Viewer.

Étape 3 : Création d’une zone de préparation (si nécessaire)¶

Créez une zone de préparation externe qui fait référence à votre compartiment GCS à l’aide de la commande CREATE STAGE. Snowflake lit vos fichiers de données en zone de préparation dans les métadonnées de la table externe. Vous pouvez aussi utiliser une zone de préparation externe.

L’exemple suivant crée une zone de préparation nommée mystage dans le schéma actif de la session utilisateur. L’URL de stockage Cloud inclut le chemin files. La zone de préparation fait référence à une intégration de stockage nommée my_storage_int.

USE SCHEMA mydb.public;

CREATE STAGE mystage
  URL='gcs://load/files/'
  STORAGE_INTEGRATION = my_storage_int;

Copy

Étape 4 : Création d’un canal avec l’intégration automatique activée¶

Créez un canal à l’aide de la commande CREATE PIPE . Le canal définit l’instruction COPY INTO <table> utilisée par Snowpipe pour charger les données de la file d’attente d’acquisition dans la table cible.

CREATE PIPE <pipe_name>
  AUTO_INGEST = true
  INTEGRATION = '<notification_integration_name>'
  AS
<copy_statement>;

Où :

<nom_canal>

Identificateur du canal ; doit être unique pour le schéma dans lequel le canal est créé.

L’identificateur doit commencer par un caractère alphabétique et ne peut pas contenir d’espaces ou de caractères spéciaux à moins que toute la chaîne d’identificateur soit délimitée par des guillemets doubles (p. ex. "My object"). Les identificateurs entre guillemets doubles sont également sensibles à la casse.

INTEGRATION = '<nom_intégration_notification>'

Nom de l’intégration de notification utilisée pour actualiser automatiquement les métadonnées de la table de répertoire à l’aide des notifications Google Cloud Pub/Sub. Une intégration de notification est un objet Snowflake qui fournit une interface entre Snowflake et des services tiers de mise en file d’attente de messages dans le Cloud.

copy_statement

Instruction COPY INTO <table> utilisée pour charger des données à partir de fichiers en file d’attente dans une table Snowflake. Cette instruction sert de texte/définition pour le canal, et elle est affichée dans la sortie SHOW PIPES.

Par exemple, créez un canal dans le schéma snowpipe_db.public qui charge les données des fichiers mis dans une zone de préparation externe (GCS) nommée mystage dans une table de destination nommée mytable :

CREATE PIPE snowpipe_db.public.mypipe
  AUTO_INGEST = true
  INTEGRATION = 'MY_NOTIFICATION_INT'
  AS
COPY INTO snowpipe_db.public.mytable
  FROM @snowpipe_db.public.mystage/path2;

Le paramètre INTEGRATION fait référence à my_notification_int l’intégration de notification créée à l” étape 1 : création d’une intégration Cloud Storage dans Snowflake. Le nom de l’intégration doit être fourni en majuscule.

Snowpipe avec l’intégration automatique est maintenant configuré !

Lorsque de nouveaux fichiers de données sont ajoutés au compartiment GCS, le message d’événement indique à Snowpipe de les charger dans la table cible définie dans le canal.

Étape 5 : Chargement de fichiers historiques¶

Pour charger les retards de traitement des fichiers de données qui existaient dans la zone de préparation externe avant que les messages Pub/Sub aient été configurés, exécutez une instruction ALTER PIPE … REFRESH.

Étape 6 : Suppression de fichiers en zone de préparation¶

Supprimez les fichiers en zone de préparation après avoir chargé avec succès les données et lorsque vous n’avez plus besoin des fichiers. Pour obtenir des instructions, voir Suppression de fichiers en zone de préparation après le chargement des données par Snowpipe.