Práticas recomendadas para Snowpipe Streaming com arquitetura de alto desempenho¶

Gerenciamento estratégico de canais¶

Aplique as seguintes estratégias de gerenciamento de canais para desempenho e estabilidade a longo prazo:

• Usar canais de longa duração: para minimizar a sobrecarga, abra um canal uma vez e mantenha-o ativo durante a tarefa de ingestão. Evite abrir e fechar canais repetidamente.

• Usar nomes de canais determinísticos: aplique uma convenção de nomenclatura consistente e previsível (por exemplo, source-env-region-client-id) para simplificar a solução de problemas e facilitar processos de recuperação automatizados.

• Escalonar com vários canais: para aumentar o rendimento, abra vários canais. Esses canais podem apontar para um único canal de destino ou para vários canais, dependendo dos limites de serviço e dos seus requisitos de taxa de transferência.

• Monitorar status do canal: use regularmente o método getChannelStatus para monitorar a integridade de seus canais de ingestão.

  • Rastreie last_committed_offset_token para verificar se os dados estão sendo ingeridos com sucesso e se o pipeline está fazendo progresso.

  • Monitore row_error_count para detectar antecipadamente registros ruins ou outros problemas de ingestão.

Validação do esquema de forma consistente¶

Certifique-se de que os dados recebidos estejam em conformidade com o esquema de tabela esperado para evitar falhas de ingestão e manter a integridade dos dados:

• Validação do lado do cliente: implemente a validação de esquema no lado do cliente para fornecer feedback imediato e reduzir erros no lado do servidor. Embora a validação completa linha por linha ofereça segurança máximo, um método com melhor desempenho pode envolver validação seletiva; por exemplo, em limites de lote ou por linhas de amostragem.

• Validação do lado do servidor: a arquitetura de alto desempenho pode descarregar a validação do esquema para o servidor. Os erros e suas contagens são relatados por meio de getChannelStatus se ocorrerem incoerências de esquema durante a ingestão no canal e na tabela de destino.

Persistência do estado para recuperação confiável¶

Para evitar perda ou duplicação de dados, seu aplicativo deve persistir em seu estado para lidar com reinicializações e falhas normalmente:

• Persistência do token de offset: após cada chamada bem-sucedida à API, faça persistir o last_committed_offset_token para armazenamento durável.

• Retomar do último ponto: ao reiniciar o aplicativo, busque o último token confirmado do Snowflake e retome a ingestão a partir desse ponto exato. Isso garante o processamento exatamente uma vez e garante a continuidade.

Adição de colunas de metadados do lado do cliente¶

Para permitir uma detecção e recuperação de erros robustas, deve-se carregar metadados de ingestão como parte da carga útil da linha. Isso requer planejamento da forma de seus dados e a definição de PIPE com antecedência.

Adicione as seguintes colunas à carga útil da linha antes da ingestão:

• CHANNEL_ID (por exemplo, um INTEGER compactado).

• STREAM_OFFSET (um BIGINT que aumenta monotonicamente por canal, como um deslocamento de partição do Kafka).

Juntas, essas colunas identificam de forma única os registros por canal e permitem que você rastreie a origem dos dados.

Opcional: adicionar uma coluna PIPE_ID se vários canais fizerem ingestão na mesma tabela de destino. Se você fizer isso, poderá rastrear as linhas facilmente de volta ao pipeline de ingestão. Você pode armazenar nomes de canal descritivos em uma tabela de pesquisa separada, mapeamento os para inteiros compactados para reduzir os custos de armazenamento.

Detecção e recuperação de erros usando offsets de metadados¶

Combine o monitoramento de canal com suas colunas de metadados para detectar e se recuperar de problemas:

• Monitorar status: verifique getChannelStatus regularmente. Um row_error_count crescente é um forte indicador de um possível problema.

• Detectar registros ausentes: se forem detectados erros, use um SQL para identificar registros ausentes ou fora de ordem, verificando lacunas em sua sequência STREAM_OFFSET.

SELECT
  PIPE_ID,
  CHANNEL_ID,
  STREAM_OFFSET,
  LAG(STREAM_OFFSET) OVER (
    PARTITION BY PIPE_ID, CHANNEL_ID
    ORDER BY STREAM_OFFSET
  ) AS previous_offset,
  (LAG(STREAM_OFFSET) OVER (
    PARTITION BY PIPE_ID, CHANNEL_ID
    ORDER BY STREAM_OFFSET
  ) + 1) AS expected_next
FROM my_table
QUALIFY STREAM_OFFSET != previous_offset + 1;

Otimização do desempenho e do custo de ingestão com MATCH_BY_COLUMN_NAME¶

Configure seu canal para mapear as colunas necessárias de seus dados de origem em vez de ingerir todos os dados em uma única coluna VARIANT. Para isso, use MATCH_BY_COLUMN_NAME = CASE_SENSITIVE ou aplique transformações na definição do canal. Essa prática recomendada não apenas otimiza seus custos de ingestão, mas também melhora o desempenho geral do pipeline de dados de streaming.

Essa prática recomendada tem as seguintes vantagens importantes:

• Ao usar MATCH_BY_COLUMN_NAME = CASE_SENSITIVE, você será cobrado apenas pelos valores de dados que são ingeridos na sua tabela de destino. Em comparação, a ingestão de dados em uma única coluna VARIANT cobra por todos os bytes JSON, tanto as chaves quanto os valores. Para dados com detalhados ou muitas chaves JSON, isso pode levar a um aumento significativo e desnecessário nos seus custos de ingestão.

• O mecanismo de processamento do Snowflake é mais eficiente em termos de computação. Em vez de analisar todo o objeto JSON em uma VARIANT, e depois extrair as colunas necessárias, esse método extrai diretamente os valores necessários.