Automação do Snowpipe para Google Cloud Storage¶

Etapa 1: Criar uma integração de armazenamento em nuvem no Snowflake¶

Crie uma integração usando o comando CREATE STORAGE INTEGRATION. Uma integração é um objeto Snowflake que delega a responsabilidade pela autenticação do armazenamento externo em nuvem a uma entidade gerada pelo Snowflake (ou seja, uma conta de serviço de armazenamento em nuvem). Para acessar os buckets de armazenamento em nuvem, o Snowflake cria uma conta de serviço que pode receber permissões para acessar os buckets que armazenam seus arquivos de dados.

Uma única integração de armazenamento pode oferecer suporte a vários estágios externos (ou seja, GCS). A URL na definição do estágio deve estar alinhada com os buckets GCS (e caminhos opcionais) especificados para o parâmetro 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>/') ]

Onde:

• integration_name é o nome da nova integração.

• bucket é o nome de um bucket de armazenamento em nuvem que armazena seus arquivos de dados (por exemplo, mybucket). O parâmetro STORAGE_ALLOWED_LOCATIONS obrigatório e o parâmetro STORAGE_BLOCKED_LOCATIONS opcional restringem ou bloqueiam o acesso a estes buckets, respectivamente, quando os estágios que fazem referência a esta integração são criados ou modificados.

• path é um caminho opcional que pode ser usado para proporcionar um controle granular sobre objetos no bucket.

O exemplo a seguir cria uma integração que limita explicitamente os estágios externos que utilizam a integração para referenciar um de dois buckets ou caminhos. Em uma etapa posterior, criaremos um estágio externo que referencia um desses buckets e caminhos.

Os estágios externos adicionais que também utilizam esta integração podem fazer referência aos buckets e caminhos permitidos:

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

Etapa 2: Recuperar a conta de serviço de armazenamento em nuvem para sua conta Snowflake¶

Execute o comando DESCRIBE INTEGRATION para recuperar a ID para a conta de serviço de armazenamento em nuvem que foi criada automaticamente para sua conta Snowflake:

DESC STORAGE INTEGRATION <integration_name>;

Onde:

• integration_name é o nome da integração que você criou em Etapa 1: Criar uma integração de armazenamento em nuvem no Snowflake (neste tópico).

Por exemplo:

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

A propriedade STORAGE_GCP_SERVICE_ACCOUNT na saída mostra a conta do serviço de armazenamento em nuvem criada para sua conta Snowflake (por exemplo, service-account-id@project1-123456.iam.gserviceaccount.com). Fornecemos uma única conta de serviço de armazenamento em nuvem para toda a sua conta Snowflake. Todas as integrações de armazenamento em nuvem utilizam essa conta de serviço.

Etapa 3: Conceder as permissões da conta de serviço para acessar os objetos do bucket¶

As seguintes instruções passo a passo descrevem como configurar as permissões de acesso IAM ao Snowflake em seu Console Google Cloud Platform para que você possa usar um bucket de armazenamento em nuvem para carregar e descarregar dados:

Pré-requisitos¶

As instruções neste tópico consideram que os seguintes itens foram criados e configurados:

Conta GCP:

• Tópico Pub/Sub que recebe mensagens de eventos do bucket GCS. Para obter mais informações, consulte Criação de tópico Pub/Sub (neste tópico).

• Assinatura que recebe mensagens de eventos do tópico Pub/Sub. Para obter mais informações, consulte Criação de assinatura Pub/Sub (neste tópico).

Para instruções, consulte a documentação Pub/Sub.

Snowflake:

• Tabela de destino no banco de dados do Snowflake onde você deseja carregar os dados.

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

Etapa 1: Criar uma integração de notificação no Snowflake¶

Crie uma integração de notificação usando o comando CREATE NOTIFICATION INTEGRATION. A integração da notificação faz referência à sua assinatura Pub/Sub. O Snowflake associa a integração da notificação com uma conta de serviço GCS criada para sua conta. O Snowflake cria uma única conta de serviço que é referenciada por todas as integrações de notificação GCS em sua conta Snowflake.

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

Onde:

• integration_name é o nome da nova integração.

• subscription_id é o nome da assinatura que você registrou em Recuperação da ID da assinatura Pub/Sub.

Por exemplo:

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

Etapa 2: Conceder acesso do Snowflake à assinatura Pub/Sub¶

1. Execute o comando DESCRIBE INTEGRATION para recuperar a ID da conta de serviço Snowflake:

   DESC NOTIFICATION INTEGRATION <integration_name>;
   

   Copy

   Onde:

   • integration_name é o nome da integração que você criou em Etapa 1: Criar uma integração de armazenamento no Snowflake.

   Por exemplo:

   DESC NOTIFICATION INTEGRATION my_notification_int;
   

   Copy

2. Registre o nome da conta de serviço na coluna GCP_PUBSUB_SERVICE_ACCOUNT, que tem o seguinte formato:

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

   Copy

3. Acesse o Console Google Cloud Platform como editor de projeto.

4. No painel inicial, escolha Big Data » Pub/Sub » Subscriptions.

5. Selecione a assinatura a ser configurada para acesso.

6. Clique em SHOW INFO PANEL no canto superior direito. O painel de informações da assinatura é exibido.

7. Clique no botão ADD PRINCIPAL.

8. No campo New principals, procure pelo nome da conta de serviço que você registrou.

9. A partir do menu suspenso Select a role, selecione Pub/Sub Subscriber.

10. Clique no botão Save. O nome da conta de serviço é adicionado ao menu suspenso da função Pub/Sub Subscriber no painel de informações.

11. Navegue até a página Dashboard no Console Cloud e selecione seu projeto na lista suspensa.

12. Clique no botão ADD PEOPLE TO THIS PROJECT.

13. Adicione o nome da conta de serviço que você registrou.

14. A partir do menu suspenso Select a role, selecione Monitoring Viewer.

15. Clique no botão Save. O nome da conta de serviço é adicionado à função Monitoring Viewer.

Etapa 3: criar um estágio (se necessário)¶

Crie um estágio externo que faça referência ao seu bucket GCS usando o comando CREATE STAGE. O Snowflake lê seus arquivos de dados preparados nos metadados da tabela externa. Você também pode usar um estágio externo já existente.

O exemplo a seguir cria um estágio chamado mystage no esquema ativo para a sessão do usuário. O URL de armazenamento em nuvem inclui o caminho files. O estágio faz referência a uma integração de armazenamento chamada my_storage_int.

USE SCHEMA mydb.public;

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

Copy

Etapa 4: criar um canal com ingestão automática habilitada¶

Crie um canal usando o comando CREATE PIPE. O canal define a instrução COPY INTO <tabela> usada pelo Snowpipe para carregar os dados da fila de ingestão na tabela de destino.

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

Onde:

<nome_do_canal>

Identificador para o canal; deve ser único para o esquema no qual o canal é criado.

O identificador deve começar com um caractere alfabético e não pode conter espaços ou caracteres especiais a menos que toda a cadeia de caracteres do identificador esteja entre aspas duplas (por exemplo, "My object"). Os identificadores delimitados por aspas duplas também diferenciam letras maiúsculas de minúsculas.

INTEGRATION = '<nome_integração_notificação>'

Nome da integração de notificação usada para atualizar automaticamente os metadados da tabela de diretório usando notificações Google Cloud Pub/Sub. Uma integração de notificação é um objeto Snowflake que fornece uma interface entre o Snowflake e serviços de enfileiramento de mensagens de terceiros.

copy_statement

Instrução COPY INTO <tabela> usada para carregar dados de arquivos em fila em uma tabela do Snowflake. Esta instrução serve como texto/definição para o canal e é exibida na saída SHOW PIPES.

Por exemplo, crie um canal no esquema snowpipe_db.public que carrega dados de arquivos preparados em um estágio externo (GCS) chamado mystage em uma tabela de destino chamada 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;

O parâmetro INTEGRATION faz referência à integração de notificação my_notification_int que você criou na Etapa 1: Criar uma integração de notificação no Snowflake. O nome da integração deve ser fornecido em letras maiúsculas.

O Snowpipe com ingestão automática está configurado!

Quando novos arquivos de dados são adicionados ao bucket GCS, a mensagem de evento informa o Snowpipe para carregá-los na tabela de destino definida no canal.

Etapa 5: carregar arquivos históricos¶

Para carregar qualquer lista de pendências de arquivos de dados que existiam no estágio externo antes que as mensagens do Pub/Sub fossem configuradas, execute uma instrução ALTERPIPE … REFRESH.

Etapa 6: excluir arquivos preparados¶

Exclua os arquivos preparados depois de carregar os dados com sucesso e não precisar mais dos arquivos. Para obter instruções, consulte Exclusão de arquivos preparados depois que o Snowpipe carrega os dados.