Snowflake na prática: entendendo Streams, Tasks e modelagem dimensional

Trabalhando com dados no Snowflake, esbarrei com um problema clássico: como manter tabelas de dimensão e fato atualizadas sem precisar reprocessar tudo todo dia? A resposta é a combinação de Streams e Tasks. Nesse artigo vou explicar cada etapa dessa combinação numa aplicação simples, dos dados ao modelo dimensional.

Em pipelines de dados, carregar a tabela inteira a cada execução é lento e custoso. O ideal é processar apenas os novos registros, inseridos, alterados ou removidos desde a última execução. Chamamos isso de Change Data Capture ou CDC, um processo que ocorre por meio das streams no Snowflake.

Stream é um objeto que rastreia as mudanças (insert, update e delete) em uma tabela. Ela não guarda uma cópia dos dados, mas sim um ponteiro sobre o histórico de versões da tabela (time travel). Um detalhe é que após as informações serem processadas para seu destino, estes são removidos da Stream.

CREATE OR REPLACE STREAM stg_pedidos_stream ON TABLE stg_pedidos
APPEND_ONLY = FALSE;

Quando a tabela stg_pedidos sofrer alguma mudança, um ponteiro será inserido na Stream.

Ao consultar a Stream, vemos as linhas alteradas mais metadados especiais:

• METADATA$ACTION: indica se a linha foi INSERT ou DELETE
• METADATA$ISUPDATE: indica se a mudança faz parte de uma atualização (nesse caso uma linha com DELETE + uma linha com INSERT)
• METADATA$ROW_ID: identificador único da linha rastreada

Para isso, podemos executar a seguinte query:

MERGE INTO fato_pedidos AS destino
USING stg_pedidos_stream AS origem
ON destino.pedido_id = origem.pedido_id
WHEN MATCHED AND origem.METADATA$ACTION = 'DELETE' THEN DELETE
WHEN MATCHED THEN UPDATE SET destino.valor = origem.valor
WHEN NOT MATCHED THEN INSERT (pedido_id, valor) VALUES (origem.pedido_id, origem.valor);

Para cada tipo de ACTION, os dados da tabela stg_pedidos são DELETADAS, ATUALIZADOS ou INSERIDOS na tabela fato_pedidos, e consumidos da Stream.

O fluxo funciona desse jeito manual, mas o grande diferencial é deixarmos tudo isso automatizado em uma Task.

Uma Task executa um comando SQL de forma agendada ou em sequência.

CREATE OR REPLACE TASK task_carrega_stg
  WAREHOUSE = COMPUTE_WH
  SCHEDULE = 'USING CRON 5 4 * * * America/Sao_Paulo'
AS
  INSERT INTO stg_pedidos
  SELECT * FROM raw_pedidos;

Nesse exemplo, nós criamos a Task task_carrega_stg, no Warehouse COMPUTE_WH, com agendamento definido pelo SCHEDULE. O que vem depois do AS é o comando que queremos executar. Você consegue ver o warehouse no canto superior direito no Workspace do Snowflake.

Pontos importantes:

• <strong>SCHEDULE </strong>aceita tanto intervalos simples ('5 MINUTE' : executa a task a cada 5 minutos) quanto expressões CRON, ideais para rodar em horários específicos, no nosso caso a Task executa as 04:05 todos os dias.
• Tasks ficam suspensas por padrão ao serem criadas, é necessário alterar seu STATE com o comando ALTER TASK ... RESUME para ativá-las
• É comum combinar Task + Stream utilizando a condição WHEN SYSTEM$STREAM_HAS_DATA(...), para garantir que a Task só rode quando realmente tiver dados para processar, evitando gastos de execução

No exemplo 0 * * * *, cada posição representa um campo: minuto, hora, dia do mês, mês e dia da semana (0 = domingo). Vale lembrar que no Snowflake o schedule CRON ainda recebe um sexto parâmetro fora dessa string, o timezone (como America/Sao_Paulo na criação da Task), que define o fuso em que os horários são interpretados.

Então, unindo a criação de Task ao nosso comando:

CREATE OR REPLACE TASK task_processa_pedidos
  WAREHOUSE = COMPUTE_WH
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('stg_pedidos_stream')
AS
  MERGE INTO fato_pedidos AS destino
  USING stg_pedidos_stream AS origem
  ON destino.pedido_id = origem.pedido_id
  WHEN MATCHED AND origem.METADATA$ACTION = 'DELETE' THEN DELETE
  WHEN MATCHED THEN UPDATE SET destino.valor = origem.valor
  WHEN NOT MATCHED THEN INSERT (pedido_id, valor) VALUES (origem.pedido_id, origem.valor);

Com isso criamos o fluxo automatizado.

Agora imagine que temos mais de uma tabela que recebe os dados brutos, antes de serem inseridos nas tabelas finais. A tabela stg_pedidos alimenta a fato_pedidos, e essa contém referências a tabela dim_cliente que por sua vez é alimentada pela tabela stg_clientes.

Esse diagrama mostra como as tabelas do exemplo se conectam:

• stg_clientes → dim_cliente: a staging de clientes alimenta a dimensão, que guarda cliente_id (chave primária ou PK – Primary Key) e nome
• stg_pedidos → fato_pedidos: a staging de pedidos alimenta a fato, que guarda pedido_id (PK), cliente_id e valor
• dim_cliente → fato_pedidos: a fato referencia a dimensão pelo cliente_id (chave estrangeira ou FK – Foreign Key) — é essa ligação que garante que cada pedido “aponte” para um cliente já existente na dimensão.

A notação ||--o{ indica a cardinalidade um-para-muitos: um cliente pode ter vários pedidos, mas cada pedido só pode ter um único cliente. E por isso a dimensão precisa ser carrega antes da tabela fato, o cliente precisa existir antes do pedido ser criado.

Nós poderíamos agendar duas Tasks com o CRON, deixando um intervalo de tempo entre as duas execuções e torcer para que a anterior já tenha terminado quando a seguinte começar. Mas isso seria ineficiente e para casos mais complexos, com várias tabelas onde cada uma processa milhares de linhas ficaria difícil garantir a ordem correta das execuções somente pelo tempo. Para isso existem as Child Tasks.

As Child Tasks formam um DAG (grafo de dependências) de execução. A Task “pai” (ou raiz) carrega o schedule, e as demais indicam a qual Task elas são dependentes.

-- Task raiz: só ela tem SCHEDULE
CREATE OR REPLACE TASK task_pai
  WAREHOUSE = COMPUTE_WH
  SCHEDULE = 'USING CRON 5 4 * * * America/Sao_Paulo'
AS
  INSERT INTO stg_pedidos SELECT * FROM raw_pedidos;

-- Child task: dispara após a task pai terminar
CREATE OR REPLACE TASK task_dim_cliente
  WAREHOUSE = COMPUTE_WH
  AFTER task_pai
  WHEN SYSTEM$STREAM_HAS_DATA('stg_clientes_stream')
AS
  MERGE INTO dim_cliente AS destino
  USING stg_clientes_stream AS origem
  ON destino.cliente_id = origem.cliente_id
  WHEN MATCHED THEN UPDATE SET destino.nome = origem.nome
  WHEN NOT MATCHED THEN INSERT (cliente_id, nome) VALUES (origem.cliente_id, origem.nome);

-- Outra child task, dependente da dimensão estar pronta
CREATE OR REPLACE TASK task_fato_pedidos
  WAREHOUSE = COMPUTE_WH
  AFTER task_dim_cliente
  WHEN SYSTEM$STREAM_HAS_DATA('stg_pedidos_stream')
AS
  MERGE INTO fato_pedidos AS destino
  USING stg_pedidos_stream AS origem
  ON destino.pedido_id = origem.pedido_id
  WHEN NOT MATCHED THEN INSERT (pedido_id, cliente_id, valor)
    VALUES (origem.pedido_id, origem.cliente_id, origem.valor);

Aqui temos a orquestração completa, a Task raiz carrega a staging, dispara a Child Task da dimensão, que por sua vez dispara a Child Task da fato, representando assim a dependência real do modelo dimensional.

Lembrando que pra ativar toda a sequência, precisamos mudar os estados de todas as Tasks para RESUME. A ordem começa pelos filhos e termina no pai (a ordem de ativação é o inverso da ordem de execução):

ALTER TASK task_fato_pedidos RESUME;
ALTER TASK task_dim_cliente RESUME;
ALTER TASK task_pai RESUME;

Você pode visualizar a árvore de dependências completa com:

SELECT *
FROM TABLE(INFORMATION_SCHEMA.TASK_DEPENDENTS(
  TASK_NAME => 'task_raiz', RECURSIVE => TRUE
));

O motivo para aplicarmos esse processo é o destino final dos dados: um modelo dimensional, onde:

• Dimensões (dim) armazenam atributos descritivos, como clientes, produtos, datas, regiões. Normalmente têm volume menor e sofrem poucas mudanças
• Fatos (fato/fact) armazenam eventos mensuráveis, como pedidos, transações, cliques, visitas. Têm alto volume e atualizações constantes

A boa prática é sempre carregar as dimensões antes dos fatos, pois normalmente a tabela fato referencia chaves geradas nas dimensões (como no exemplo de fato_pedidos referenciar dim_cliente).

Streams, Tasks e Child Tasks com agendamento, resolvem juntos o problema de manter um Data Warehouse dimensional atualizado de forma barata e ordenada. O modelo Fato-Dimensão fica sempre atualizado, processando apenas o que mudou, com um custo proporcional ao volume real das mudanças, e não o volume total das tabelas.

É uma arquitetura simples de aplicar no Snowflake para pipelines de pequeno a médio porte, economizando créditos e eliminando boa parte da complexidade de orquestradores externos.

Imagem de capa por Marcin Zukowski na Snowflake.