Resiliência multilocal para pipelines de dados¶

Requisitos¶

• Edição: Snowflake Business Critical Edition (ou superior).

• Métodos de ingestão compatíveis: esse recurso é exclusivamente compatível com o carregamento de dados baseado em arquivos por meio do Snowpipe (ingestão automática) e COPY INTO <table>. Ele não é compatível com o Openflow ou o Snowpipe Streaming.

• Estruturas de caminho idênticas:** para permitir que seus pipelines localizem novos arquivos após o failover, você deve gravá-los no local de armazenamento secundário usando exatamente a mesma hierarquia, estrutura de pastas e caminho relativo do seu local primário.

Considerações¶

• Faturamento:** esse recurso incorre em cobranças de replicação padrão (transferência de dados e recursos de computação), faturadas em sua conta de destino.

• Tempo de inatividade para modificação de área de preparação: alterar a propriedade RELATIVE_URL em uma área de preparação existente invalidará os objetos dependentes e interromperá a ingestão. Recomendamos a criação de novas áreas de preparação durante a configuração para evitar tempo de inatividade.

• Integração de notificação de várias filas (MQNI):** o uso da mesma fila ativa nas contas de origem e destino não é compatível. Fazer isso pode resultar em perda de notificações. O Snowflake não verifica se a mesma fila está em uso em todas as contas.

• Tabela de diretórios: a criação de uma tabela de diretórios em uma área de preparação usando a MLSI não é compatível atualmente.

Comportamento de replicação¶

• Replicação assíncrona: o Snowflake replica suas tabelas e o histórico de carga do seu pipeline simultaneamente no mesmo instantâneo. Como eles estão sincronizados, uma interrupção não resultará em dados duplicados. Se o seu banco de dados secundário estiver com quatro horas de atraso, os dados da tabela também estarão com quatro horas de atraso, e o processamento de quatro horas de notificações enfileiradas simplesmente atualiza a tabela.

• Evitar perda de dados em gravação dupla: seu objetivo de ponto de recuperação (Recovery Point Objective, RPO) é ditado pelo seu intervalo de atualização de replicação. Para evitar perda de dados durante um failover, o período de retenção de mensagens da sua fila de mensagens na nuvem secundária deve ser maior que o seu intervalo de replicação. Se a sua fila descartar mensagens antes que a replicação agendada seja concluída, esses arquivos não serão ingeridos durante o failover.

• Risco de perda de dados em gravação única: se você usar roteamento de gravação única, todos os arquivos processados ​​no local primário após a última replicação bem-sucedida serão totalmente desconhecidos para o local secundário. Em caso de failover, esses dados ficarão temporariamente ausentes na sua conta de destino.

1. Gravações duplas – recomendado¶

Seu aplicativo produtor grava arquivos nos buckets de armazenamento em nuvem primário e secundário simultaneamente. A fila de mensagens secundária acumula notificações, mas o Snowflake não as processa porque o banco de dados secundário é somente leitura.

• O que acontece em caso de failover: o banco de dados secundário torna-se gravável. O Snowpipe lê a fila secundária e usa o histórico de carregamento replicado para remover arquivos duplicados. Se uma interrupção impediu que um arquivo fosse concluído no local primário, o pipeline secundário lê a notificação da fila secundária, vê que o arquivo está ausente do histórico de carregamento e o ingere.

• O que acontece em caso de failback: quando o local primário se recupera e você atualiza o grupo de failover e, em seguida, realiza o failback, o Snowpipe começa automaticamente a ingerir novos arquivos, já que o histórico de carregamento foi sincronizado da conta secundária durante a preparação do failback.

• Resultado:** sem dados ausentes nem duplicados. O Snowflake lida com a reconciliação automaticamente em ambas as direções.

• Ação necessária:** nenhuma, além da sincronização de replicação padrão pré-failback (ALTER FAILOVER GROUP … REFRESH) para garantir que a conta primária tenha o histórico de carregamento mais recente.

2. Roteamento de gravação única¶

Seu produtor grava apenas no armazenamento em nuvem primário. Em caso de interrupção, você redireciona o produtor para começar a gravar novos arquivos no armazenamento em nuvem secundário.

• O que acontece no failover: a conta secundária começa imediatamente a processar novos arquivos roteados para o bucket secundário. No entanto, todos os arquivos em trânsito que estejam retidos no local primário afetado são deixados para trás temporariamente.

• O que acontece no failback: quando o local primário se recupera e você retorna à sua conta Snowflake primária, o Snowpipe processa automaticamente todas as notificações de arquivos que chegaram à fila com sucesso antes da interrupção.

• Resultado: sem duplicatas. No entanto, os arquivos em que a notificação na nuvem falhou completamente em ser gerada devido à interrupção (ou em que a interrupção durou mais do que a política de retenção de mensagens da sua fila) exigem intervenção manual.

• Ação necessária: após o failback, compare seu bucket de armazenamento primário com a exibição COPY_HISTORY no Snowflake para identificar quaisquer arquivos ausentes. Execute ALTER PIPE … REFRESH ou um comando manual COPY INTO para carregar esses arquivos específicos que ficaram retidos.

Etapa 1: Criar uma integração de armazenamento multilocal (MLSI)¶

Para configurar uma integração de armazenamento multilocal, você segue as etapas padrão para configurar uma integração de armazenamento com algumas diferenças observadas nesta seção.

Em sua conta de origem, crie a MLSI fornecendo valores para cada local na lista STORAGE_LOCATIONS. Você pode combinar provedores de nuvem para configurações entre nuvens.

CREATE STORAGE INTEGRATION my_mlsi
  TYPE = EXTERNAL_STAGE
  STORAGE_LOCATIONS =
  (
    (
      NAME = 'my-s3-us-west-1'
      STORAGE_PROVIDER = 'S3'
      STORAGE_BASE_URL = 's3://myBucketWest'
      STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::12345:role/myrole'
      STORAGE_AWS_EXTERNAL_ID = 'mlsi-external-id'
      ENCRYPTION = ( TYPE = 'AWS_SSE_S3' )
    ),
    (
      NAME = 'my-s3-us-east-1'
      STORAGE_PROVIDER = 'S3'
      STORAGE_BASE_URL = 's3://myBucketEast'
      STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::67890:role/myrole'
      STORAGE_AWS_EXTERNAL_ID = 'mlsi-external-id'
      ENCRYPTION = ( TYPE = 'AWS_SSE_S3' )
    )
  )
  ENABLED = TRUE
  STORAGE_ALLOWED_LOCATIONS = ('*')
  ACTIVE = 'my-s3-us-west-1';

Onde:

• STORAGE_LOCATIONS: especifica uma lista de um ou mais locais de armazenamento (bucket S3, bucket GCS ou contêiner do Azure) para a integração de armazenamento. Para visualizar os parâmetros de cada provedor de nuvem, consulte parâmetros do provedor de nuvem (cloudProviderParams) na página de referência CREATE STORAGE INTEGRATION.

• NAME: cadeia de caracteres que especifica o identificador (nome) do local de armazenamento.

• ENCRYPTION: especifica a criptografia para o local de armazenamento. Você deve especificar a criptografia para o local de armazenamento no nível da integração de armazenamento, em vez de no nível da área de preparação. Necessário apenas para carregamento de arquivos criptografados; não é necessário se o local de armazenamento e os arquivos não estão criptografados. Para visualizar as opções de criptografia para cada provedor de nuvem, consulte ENCRYPTION na página de referência CREATE STAGE.

• ACTIVE: especifica o nome do local de armazenamento a ser definido como o local ativo para a integração de armazenamento na conta atual.

  Para o local de armazenamento ativo em sua conta de origem, você deve configurar o controle de acesso e conceder permissão ao Snowflake para acessar seu armazenamento. Siga as instruções nos seguintes tópicos:

  • Configurando uma integração de armazenamento do Snowflake para acessar o Amazon S3

  • Configurar uma integração para o Google Cloud Storage

  • Configurar um contêiner do Azure para carregamento de dados

Etapa 2: Associar a MLSI a uma área de preparação externa¶

Recomendamos a criação de uma nova área de preparação em vez de alterar uma existente.

Use o comando CREATE STAGE para associar a integração de armazenamento multilocal que você criou a uma ou mais áreas de preparação externas:

CREATE STAGE my_ext_stage
  RELATIVE_URL = '/my_folder/my_sub_folder/'
  STORAGE_INTEGRATION = 'my_mlsi';

Onde:

• RELATIVE_URL: o caminho relativo para o local da sua área de preparação externa a partir do local de armazenamento definido em sua integração de armazenamento. Para permitir que seus pipelines localizem novos arquivos após um failover, você deve gravá-los no local de armazenamento secundário usando a mesma hierarquia, estrutura de pastas e caminho relativo do seu local primário.

• STORAGE_INTEGRATION: o nome da sua integração de armazenamento multilocal.

Por exemplo:

ALTER STAGE my_ext_stage SET
  RELATIVE_URL = '/my_folder/my_sub_folder/'
  STORAGE_INTEGRATION = 'my_mlsi';

Etapa 3: Configurar uma integração de notificação de várias filas (MQNI)¶

Se você usa carregamento de dados automatizado por meio de mensagens na nuvem e configurou uma integração de armazenamento multilocal para a sua área de preparação externa, também deve usar uma integração de notificação de várias filas para failover contínuo dos seus pipelines do Snowpipe.

Para cada fila que você definir para a integração de notificações, você deve preparar seu serviço de mensagens seguindo as etapas descritas nos tópicos a seguir:

• Configurando o Amazon SNS para automatizar o Snowpipe usando notificações SQS. Crie um tópico SNS para cada região AWS na qual sua MLSI tenha locais de armazenamento.

• Configurando a automação usando o GCS Pub/Sub

• Configurando a automação com o Azure Event Grid

CREATE NOTIFICATION INTEGRATION my_mqni
  ENABLED = TRUE
  TYPE = MULTI_QUEUE
  DIRECTION = INBOUND
  QUEUES = (
    (
      NAME = 'my-us-west-1'
      NOTIFICATION_PROVIDER = AWS_SNS
      AWS_SNS_TOPIC_ARN = 'arn:aws:sns:us-west-1:12345:my-snowpipe-mlsi-west'
    ),
    (
      NAME = 'my-us-east-1'
      NOTIFICATION_PROVIDER = AWS_SNS
      AWS_SNS_TOPIC_ARN = 'arn:aws:sns:us-west-1:12345:my-snowpipe-mlsi-east'
    )
  )
  ACTIVE = 'my-us-west-1';

CREATE PIPE my_pipe
  AUTO_INGEST = TRUE
  INTEGRATION = my_mqni
  AS COPY INTO my_table FROM @my_ext_stage/my_pipe/;

SYSTEM$CONVERT_PIPES_TO_MULTI_QUEUE(
  '<new_mqni_name>',
  '<original_sns_topic_arn_or_int_name>',
  '<new_sns_topic_arn_or_int_name>'
)

SELECT SYSTEM$CONVERT_PIPES_TO_MULTI_QUEUE(
  'my_mqni',
  'arn:aws:sns:us-west-1:12345:my-snowpipe-mlsi-west',
  'arn:aws:sns:us-east-1:67890:my-snowpipe-mlsi-east'
);

SELECT SYSTEM$CONVERT_PIPES_TO_MULTI_QUEUE(
  'my_azure_mqni',
  'my_azure_ni_1',
  'my_azure_ni_2'
);

Etapa 4: Replicar suas MLSI e MQNI para a conta de destino.¶

1. To replicate your multi-location storage integration and multi-queue notification integration, alter your existing replication or failover group to include STORAGE INTEGRATIONS and NOTIFICATION INTEGRATIONS in the ALLOWED_INTEGRATION_TYPES list.

Por exemplo, use o comando ALTER FAILOVER GROUP:

ALTER FAILOVER GROUP my_fg SET
  OBJECT_TYPES = DATABASES, ROLES, INTEGRATIONS
  ALLOWED_INTEGRATION_TYPES = API INTEGRATIONS, STORAGE INTEGRATIONS,
    NOTIFICATION INTEGRATIONS;

2. Em seguida, em sua conta de destino, execute uma operação de atualização:

ALTER FAILOVER GROUP my_fg REFRESH;

Etapa 5: Configurar os estados ativos na conta de destino¶

Após executar uma operação de atualização, para garantir um failover sem problemas durante uma interrupção real, configure o local de armazenamento ativo e a fila (se estiver usando uma integração de notificação) em sua conta de destino.

Em sua conta de destino:

1. Para o local de armazenamento que você deseja definir como o local ativo na sua conta de destino, use as instruções nos tópicos a seguir para conceder ao Snowflake acesso ao seu armazenamento:

   • Opção 1: Configurar uma integração de armazenamento do Snowflake para acessar o Amazon S3

   • Configurar uma integração para o Google Cloud Storage

   • Configurar um contêiner do Azure para carregamento de dados

2. Ative o armazenamento secundário: defina a MLSI para usar seu local de armazenamento de backup secundário na conta de destino.

   CopyExpand

   ALTER STORAGE INTEGRATION my_mlsi SET ACTIVE = 'my-s3-us-east-1';
   

   Show lessSee more

   Scroll to top

3. Se você estiver usando uma Integração de notificação de várias filas, conceda permissão ao Snowflake para acessar seu serviço de mensagens para a fila que você deseja definir como ativa na sua conta de destino. Siga as instruções para seu provedor de nuvem:

   • Para a AWS, consulte Inscrever a fila SQS do Snowflake no tópico SNS em Pré-requisito: Criar um tópico e assinatura do Amazon SNS.

   • Para o Google Cloud, consulte Conceder acesso do Snowflake à assinatura do Pub/Sub em Configuração da automação usando GCS Pub/Sub.

   • Para o Azure, consulte Conceder acesso do Snowflake à fila de armazenamento em Configuração da automação com a Event Grid do Azure.

4. Ative a fila secundária (se estiver usando MQNI): defina a fila ativa para seu local secundário na conta de destino.

   CopyExpand

   ALTER INTEGRATION my_mqni
     SET ACTIVE = 'MY_MQNI-queue2';
   

   Show lessSee more

   Scroll to top

1. Verificar os estados de integração ativos¶

Confirme se as suas integrações estão apontando para o armazenamento e as filas corretas, verificando as propriedades delas. Procure a propriedade ACTIVE na saída:

-- Check the active storage location
DESCRIBE STORAGE INTEGRATION my_mlsi;

-- Check the active message queue
DESCRIBE INTEGRATION my_mqni;

2. Verificar o status do canal (somente Snowpipe)¶

Use a função SYSTEM$PIPE_STATUS para garantir que seu canal esteja em execução e para verificar se ele está enfileirando ativamente novos arquivos do seu local secundário.

SELECT SYSTEM$PIPE_STATUS('my_pipe');

Procure por «executionState»:»RUNNING» e verifique «pendingFileCount» para confirmar se ele está reconhecendo ativamente novos arquivos adicionados ao seu bucket secundário.

3. Validar a ingestão bem-sucedida (histórico de carregamento)¶

Para garantir que os dados estejam sendo carregados sem erros ou duplicados, consulte a exibição COPY_HISTORY. Ela mostra exatamente quais arquivos foram ingeridos, o caminho de origem deles e quando foram carregados.

SELECT file_name, status, row_count, last_load_time
FROM TABLE(information_schema.copy_history(
  table_name => 'my_table',
  start_time => DATEADD(hours, -1, CURRENT_TIMESTAMP())
));

Verifique se os caminhos de file_name refletem seu local de armazenamento ativo e se o status é exibido como LOADED.