Replicação de histórico de carregamento, canal e estágio¶

Requisitos¶

Antes de replicar objetos de pipeline de dados, você deve ativar a replicação ETL. O suporte para replicação ETL faz parte do pacote 2024_01 BCR. Você pode ativar o pacote 2024_01 BCR executando a seguinte instrução:

SELECT SYSTEM$ENABLE_BEHAVIOR_CHANGE_BUNDLE('2024_01');

Para replicar quaisquer estágios externos que usam uma integração de armazenamento, você também deve configurar seu grupo de replicação ou failover para replicar STORAGE INTEGRATIONS. Caso contrário, os estágios externos serão replicados sem a integração de armazenamento associada.

Você pode usar uma instrução ALTER REPLICATION GROUP ou ALTER FAILOVER GROUP para modificar essas propriedades de um grupo existente.

Por exemplo:

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

Replicação e estágios¶

Esta seção descreve o nível atual de funcionalidade de replicação compatível com o Snowflake para diferentes tipos de estágios.

Replicação e canais¶

Esta seção descreve o nível atual de funcionalidade de replicação com suporte para diferentes tipos de canais.

Snowflake oferece suporte à replicação para o seguinte:

• Objetos de canal, incluindo ingestão automática e canais do ponto de extremidade REST que carregam dados de estágios externos.

• Parâmetros de nível de canal.

• Concessões de privilégios em objetos de canal.

Exemplo 1: replicar um estágio interno nomeado¶

Este exemplo demonstra como funciona a replicação para estágios internos. Em especial, o exemplo mostra como a tabela de diretórios é a única fonte de verdade para os metadados da tabela antes e depois da replicação.

A primeira parte do exemplo conclui as tarefas a seguir em uma conta de origem.

1. Crie um estágio interno chamado my_int_stage com uma tabela de diretórios habilitada para replicar os arquivos no estágio. Em seguida, copie os dados de uma tabela chamada my_table para os arquivos do estágio.

   Nota

   O exemplo atualiza a tabela de diretórios após carregar file1 e file2 no estágio para sincronizar os metadados da tabela com o conjunto mais recente de arquivos na definição do estágio para as tabelas de diretórios. No entanto, nenhuma operação de atualização ocorre após o carregamento 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. Criação de um grupo de failover:

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

   Copy

A segunda parte do exemplo conclui o processo de replicação e failover em uma conta de destino:

1. Crie um grupo de failover como uma réplica do grupo de failover na conta de origem, atualize os objetos no novo grupo de failover e promova a conta de destino para servir como conta de origem.

   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. Em seguida, atualize a tabela de diretórios no estágio replicado e copie todos os arquivos rastreados pela tabela de diretórios em my_stage para uma tabela chamada my_table.

   Nota

   A instrução COPY INTO carrega file1 e file2 na tabela, mas não file3. Isso ocorre porque a tabela de diretórios não foi atualizada após adicionar file3 na conta de origem.

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

   Copy

Exemplo 2: replicar um estágio externo e integração de armazenamento¶

Este exemplo fornece um fluxo de trabalho de amostra para replicar um estágio externo e integração de armazenamento para uma conta de destino.

O exemplo pressupõe que você já tenha concluído o seguinte: Configuração do acesso seguro ao seu bucket Amazon S3.

A primeira parte do exemplo conclui as tarefas a seguir em uma conta de origem.

1. Crie uma integração de armazenamento para um bucket Amazon S3 no banco de dados 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. Crie um estágio externo no banco de dados my_database_2 usando a integração de armazenamento my_storage_int.

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

   Copy

3. Crie um grupo de failover e inclua o banco de dados my_database_2 e objetos de integração de armazenamento.

   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

A segunda parte do exemplo conclui o processo de replicação e failover em uma conta de destino:

1. Crie um grupo de failover como uma réplica do grupo de failover na conta de origem e atualize.

   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. Depois de replicar a integração de armazenamento para a conta de destino, você deverá executar etapas adicionais para atualizar as permissões do seu provedor de nuvem para conceder acesso ao seu armazenamento em nuvem à integração de replicação. Para obter mais informações, consulte Configuração do acesso ao armazenamento em nuvem para integrações de armazenamento secundárias.

Exemplo 3: replicar um canal de ingestão automática¶

Este exemplo fornece um fluxo de trabalho de amostra para replicar um canal que usa um tópico do Amazon Simple Notification Service (SNS) com o Amazon Simple Queue Service (SQS) para automatizar o Snowpipe.

O exemplo pressupõe que você já tenha concluído as seguintes tarefas:

• Criação e configuração de uma integração de armazenamento para Amazon S3. Por exemplo, usamos uma integração de armazenamento chamada my_s3_storage_int.

• Criação de um tópico e uma assinatura SNS da Amazon e inscrição na fila Snowflake SQS em seu tópico SNS.

• Criação de um estágio externo que faz referência à sua integração de armazenamento. Para fins de amostra, usamos um estágio chamado my_s3_stage. Para obter instruções, consulte CREATE STAGE.

Comece com as tarefas a seguir em uma conta de origem.

1. Use o comando CREATE PIPE para criar um canal com ingestão automática habilitada que carregue dados do estágio externo em uma tabela chamada 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. Atualize o canal com uma instrução ALTER PIPE para carregar dados do estágio dos últimos 7 dias.

   ALTER PIPE mypipe REFRESH;
   

   Copy

3. Por fim, use CREATE FAILOVER GROUP para criar um grupo de failover que permita a replicação de integrações de armazenamento.

   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

A segunda parte do exemplo conclui o processo de replicação e failover em uma conta de destino:

1. Crie um grupo de failover como uma réplica do grupo de failover na conta de origem:

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

   Copy

2. Execute uma instrução DESCRIBE INTEGRATION para recuperar o ARN do usuário AWS IAM da sua conta Snowflake na implantação secundária.

   Use ARN para conceder ao usuário IAM permissões para acessar seu bucket S3. Consulte a Etapa 5: conceder ao usuário IAM permissões de usuário para acessar objetos de bucket.

   DESC INTEGRATION my_s3_storage_int;
   

   Copy

3. Chame a função do sistema SYSTEM$GET_AWS_SNS_IAM_POLICY para gerar uma política IAM que conceda à nova fila SQS permissão para assinar seu tópico SNS. O Snowflake criou a nova fila SQS na sua conta de destino quando você replicou o grupo de failover da sua conta de origem.

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

   Copy

   topic_arn é o Amazon Resource Name (ARN) do tópico SNS que você criou para o canal original na sua conta de origem.

   Em seguida, Inscreva-se na nova fila da Amazon SQS em seu tópico SNS.

4. Atualize os objetos no seu novo grupo de failover.

   ALTER FAILOVER GROUP my_pipe_failover_group REFRESH;
   

   Copy

5. Por fim, promova a conta de destino para servir como conta de origem com o comando ALTER FAILOVER GROUP.

   ALTER FAILOVER GROUP my_pipe_failover_group PRIMARY;
   

   Copy

   O canal mypipe começará a carregar todos os dados disponibilizados desde a última vez que o grupo de failover foi atualizado na conta de origem.

   Para verificar se o canal replicado está funcionando, consulte a tabela da instrução COPY do canal.

   SELECT * FROM mytable;
   

   Copy