Monitore o Snowpipe Streaming usando tabelas de eventos

Monitore pipelines do Snowpipe Streaming diretamente no Snowflake para verificar se os dados estão chegando, identificar atrasos de processamento e investigar falhas de ingestão usando tabelas de eventos. O Snowflake fornece uma tabela de eventos padrão para coletar esses dados, que registram métricas como latência, vazão, contagem de linhas e erros em canais.

Para que serve

Imagine que você precisa garantir que os dados de streaming estão entrando nas tabelas sem atrasos e quer criar alertas para falhas de gravação. Com esse recurso você consegue:

• Acompanhar contagens de linhas processadas e rejeitadas em tempo real.
• Medir a latência do lado do servidor entre o recebimento e o commit dos dados.
• Investigar erros de linhas e falhas em canais específicos.
• Construir dashboards e configurar alertas usando consultas SQL.

Como usar

Primeiro, verifique se a coleta de eventos está ativa ajustando o parâmetro LOG_EVENT_LEVEL para INFO no schema que contém suas tabelas de destino:

ALTER SCHEMA meu_banco.meu_schema SET LOG_EVENT_LEVEL = INFO;

O que acontece nesse código:

1. Altera o nível de log de eventos do schema para INFO.
2. Habilita a gravação de todos os tipos de eventos do Snowpipe Streaming na tabela de eventos configurada.

Em seguida, consulte os eventos recentes de commit para verificar a vazão e o progresso da ingestão:

SELECT
  timestamp,
  resource_attributes:"snow.table.name"::STRING AS table_name,
  value:"channel_name"::STRING AS channel_name,
  value:"row_count"::NUMBER AS rows_ingested,
  value:"error_count"::NUMBER AS error_count
FROM SNOWFLAKE.TELEMETRY.EVENTS_VIEW
WHERE scope:"name"::STRING = 'snow.snowpipe.streaming'
  AND record_type = 'EVENT'
  AND record:"name"::STRING = 'commit'
  AND resource_attributes:"snow.database.name"::STRING = 'meu_banco'
  AND resource_attributes:"snow.schema.name"::STRING = 'meu_schema'
  AND resource_attributes:"snow.table.name"::STRING = 'minha_tabela'
  AND timestamp > DATEADD('minute', -15, CURRENT_TIMESTAMP())
ORDER BY timestamp DESC;

O que acontece nesse código:

1. Filtra os registros pelo escopo específico do Snowpipe Streaming e pelo tipo de evento de commit.
2. Restringe a busca ao banco, schema e tabela de destino informados.
3. Retorna os dados dos últimos 15 minutos ordenados por timestamp decrescente.

Limitações e requisitos

• Exige o uso de um pipeline de Snowpipe Streaming com arquitetura de alta performance.
• É necessário ter uma tabela de eventos ativa e uma role com permissão de consulta (como a role SNOWFLAKE.EVENTS_VIEWER para a tabela padrão).
• O nível LOG_EVENT_LEVEL deve estar configurado como INFO ou ERROR no schema de destino.