ETL no Databricks SQL

Ao lidar com grandes quantidades de dados, você precisa de um pipeline que possa processar apenas os registros novos e alterados, em vez de reprocessar todo o dataset. Isso se chama ETL incremental. No Databricks SQL, você pode criar um pipeline ETL incremental usando tabelas de transmissão e visualizações materializadas, sem escrever código procedural ou realizar atualizações manuais programadas.

Este tutorial mostra um padrão comum: acompanhamento de alterações de produto ao longo do tempo. Você cria uma tabela de origem, captura eventos de alteração, cria uma tabela de dimensões que preserva a história completa de cada produto, mantém uma tabela de fatos que sobrescreve registros recentes incrementalmente e adiciona uma camada de relatório agregado na parte superior. Você usa AUTO CDC para a tabela de dimensão e um fluxo REPLACE WHERE para a tabela de fatos.

O recurso key neste tutorial é AUTO CDC. Num armazém tradicional, você escreveria instruções MERGE INTO complexas para conciliar eventos de inserção, atualização e exclusão em uma tabela de destino. Essa abordagem é propensa a erros, especialmente quando os eventos chegam fora de ordem. AUTO CDC trata disso para você. Você declara a key de negócio, a coluna de sequenciamento e se deseja SCD Tipo 1 (somente o valor mais recente) ou SCD Tipo 2 (histórico completo), e Databricks aplica automaticamente a lógica merge correta. Para obter uma visão geral do CDC, consulte APIs do AUTO CDC : Simplifique a captura de dados de alterações (CDC) com pipeline.

Ao final deste tutorial, você terá:

1. Criei uma tabela de origem que rastreia alterações com o Feed de Dados de Alteração.
2. Analisei os dados brutos de alterações para compreender a transmissão de eventos CDC .
3. Utilizou-se AUTO CDC para construir uma tabela de dimensão SCD Tipo 2 a partir desses eventos.
4. Os eventos de exclusão foram processados incrementalmente ao longo do pipeline.
5. Criou uma tabela de fatos que substitui gradualmente as linhas recentes por um fluxo REPLACE WHERE.
6. Criou-se uma view materializada que mantém incrementalmente um relatório agregado.
7. Configurado SCHEDULE REFRESH EVERY 1 DAY para que as alterações se propaguem automaticamente pelo pipeline.

Requisitos

Para completar este tutorial, você deve atender aos seguintes requisitos:

• Um workspace Databricks com Unity Catalog ativado.
• Um SQL warehouse (serverless ou pro).
• Ter permissão para criar um recurso compute ou acesso a um recurso compute.
• computesem servidor habilitada para sua account. Veja recurso com disponibilidade regional limitada.

Passo 1: Configure seu catálogo e esquema

Abra o editorDatabricks SQL e defina seu catálogo e esquema de trabalho. Você precisa ter permissão para USE o catálogo e o esquema que você selecionar:

SQL

USE CATALOG <your-catalog>;
USE SCHEMA <your-schema>;

Etapa 2: Criar uma tabela de origem e carregar dados

Crie uma tabela products com Usar a alteração do feed de dados no Databricks (CDF) ativado. O CDF é um recurso do Delta Lake que registra cada inserção, atualização e exclusão como um log de alterações consultável. Isso é similar a uma transmissão CDC de um sistema de origem transacional, exceto que as mudanças são capturadas diretamente na tabela Delta em vez de um log externo. O CDF é usado aqui para gerar os eventos de alteração que o pipeline downstream consumirá.

1. Crie a tabela e carregue os registros iniciais:

   SQL

   CREATE OR REPLACE TABLE products (
     product_id INT,
     product_name STRING,
     category STRING,
     warehouse STRING
   )
   TBLPROPERTIES (delta.enableChangeDataFeed = true);
   
   INSERT INTO products VALUES
     (1, 'Spoon', 'Cutlery', 'Seattle'),
     (2, 'Fork', 'Cutlery', 'Portland'),
     (3, 'Knife', 'Cutlery', 'Denver'),
     (4, 'Chair', 'Furniture', 'Austin'),
     (5, 'Table', 'Furniture', 'Chicago'),
     (6, 'Lamp', 'Lighting', 'Boston'),
     (7, 'Mug', 'Kitchenware', 'Seattle'),
     (8, 'Plate', 'Kitchenware', 'Atlanta'),
     (9, 'Bowl', 'Kitchenware', 'Dallas'),
     (10, 'Glass', 'Kitchenware', 'Phoenix');

2. Simule alterações a montante, incluindo novos produtos, mudança de armazém e reatribuição de categoria:

   SQL

   INSERT INTO products VALUES
     (11, 'Napkin', 'Dining', 'San Francisco'),
     (12, 'Coaster', 'Dining', 'New York');
   
   UPDATE products SET warehouse = 'Los Angeles' WHERE product_id = 1;
   UPDATE products SET category = 'Dining' WHERE product_id = 2;

o passo 3: Consultar o feed de dados alterados

Antes de construir o pipeline downstream, é útil analisar os eventos de mudança brutos para que você possa entender o que AUTO CDC irá processar. A função table_changes() lê o log do CDF e retorna todas as operações capturadas juntamente com as colunas de metadados:

SQL

SELECT
  product_id, product_name, warehouse,
  _change_type, _commit_version
FROM table_changes('products', 1)
ORDER BY _commit_version, product_id;

Por exemplo, a Colher tem três eventos: um insert (Seattle), um update_preimage (Seattle) e um update_postimage (Los Angeles).

Note que uma única mudança lógica (por exemplo, mover a colher para um armazém diferente) produz múltiplos eventos: uma pré-imagem e uma pós-imagem. Num data warehouse tradicional, você escreveria uma instrução MERGE para reconciliar todos esses eventos em uma tabela de destino, lidando com inserções, atualizações e exclusões com lógica separada e garantindo que os eventos sejam aplicados na ordem correta. Esta é exatamente a complexidade que AUTO CDC elimina no próximo passo.

o passo 4: Construir uma dimensão SCD Tipo 2 com AUTO CDC

Uma tabela de transmissão processa dados de forma incremental. A cada refresh, o programa lê apenas as novas linhas desde a última execução, portanto não precisa reprocessar todo o dataset. Isso o torna ideal para fontes de alto volume ou que mudam com frequência.

AUTO CDC adiciona processamento de captura de dados de alterações (CDC) em cima de uma tabela de transmissão. Em vez de escrever uma instrução MERGE INTO que lida manualmente com inserções, atualizações e exclusões, você declara a key de negócio e a coluna de sequenciamento e deixa Databricks aplicar a lógica correta. AUTO CDC também lida com eventos fora de ordem automaticamente, o que é um problema comum ao usar MERGE INTO para lidar com eventos que chegam de sistemas distribuídos ou cargas de lotes com carimbos de data/hora sobrepostos.

A seguinte instrução cria uma tabela SCD Tipo 2 que preserva o histórico completo de versões de cada produto. Cada versão recebe carimbos de data/hora __START_AT e __END_AT . Um NULL em __END_AT marca a versão atual.

SQL

CREATE OR REFRESH STREAMING TABLE products_history
SCHEDULE REFRESH EVERY 1 DAY
FLOW AUTO CDC
FROM (
  SELECT * FROM STREAM products WITH (readChangeFeed = true)
  WHERE _change_type != 'update_preimage'
)
KEYS (product_id)
APPLY AS DELETE WHEN _change_type = 'delete'
SEQUENCE BY _commit_timestamp
COLUMNS * EXCEPT (_change_type, _commit_version, _commit_timestamp)
STORED AS SCD TYPE 2;

• SCHEDULE REFRESH EVERY 1 DAYAtualizar a tabela diariamente, conforme programado.
• FLOW AUTO CDC: declara isso como um fluxo do CDC. O Databricks aplica automaticamente as semânticas de inserção, atualização e exclusão.
• WHERE _change_type != 'update_preimage': ignora pré-imagens. A pós-imagem já possui a linha atualizada.
• KEYS (product_id): a key do negócio. Eventos com a mesma key são mesclados em linhas versionadas.
• APPLY AS DELETE WHEN _change_type = 'delete': Encerra a versão atual quando um evento de exclusão chega. Isso permite definir a condição que identifica um evento de exclusão.
• SEQUENCE BY _commit_timestamp: estabelece a ordem dos eventos. Lida corretamente com chegadas fora de ordem.
• STORED AS SCD TYPE 2: mantém a história completa. AUTO CDC suporta tanto SCD Tipo 1 quanto SCD Tipo 2.

Consultar a tabela de dimensões:

SQL

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;

• Colher: duas versões. Seattle (fechado, conjunto __END_AT ) e Los Angeles (atual, __END_AT = NULL).
• Bifurcação: duas versões. Categoria Talheres (fechada) e Categoria Refeições (atual).
• Guardanapo e porta-copos: uma versão de cada (recém-inseridos, __END_AT = NULL).
• Todos os outros produtos: uma versão cada (__END_AT = NULL).

o passo 5: Processar exclusões através do pipeline

Agora, simule dois produtos descontinuados, excluindo-os da tabela de origem:

SQL

DELETE FROM products WHERE product_id = 9;
DELETE FROM products WHERE product_id = 10;

Esses eventos de exclusão são registrados no log do CDF, mas a tabela de transmissão ainda não os viu. Atualize a tabela de transmissão para processar os novos eventos:

SQL

REFRESH STREAMING TABLE products_history;

Consulte a tabela de dimensões para verificar se as exclusões foram aplicadas:

SQL

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;

O Bowl e o Glass agora estão fechados com __END_AT definido, marcando-os como descontinuados. Todos os demais produtos atuais permanecem inalterados. A tabela de transmissão processou apenas os novos eventos de exclusão, sem reprocessar as inserções e atualizações da refresh anterior.

O passo 6: Construir uma tabela de fatos com um fluxo REPLACE WHERE

A tabela de dimensão nas passos anteriores usa AUTO CDC porque rastreia o estado atual e histórico de cada produto. Tabelas de fatos têm necessidades diferentes. Uma tabela de fatos registra eventos, como ventas, e você normalmente reprocessa apenas uma janela recente, por exemplo, para corrigir registros que chegam com atraso. Para este padrão, use um fluxo REPLACE WHERE incremental, que sobrescreve apenas as linhas que correspondem a um predicado e deixa todas as outras linhas intocadas. Para obter uma comparação completa desta construção com visualizações materializadas, consulte Quando usar REPLACE WHERE x visualizações materializadas.

1. Crie uma tabela product_sales e carregue alguns eventos de ventas, alguns recentes e outros mais antigos:

   SQL

   CREATE OR REPLACE TABLE product_sales (
     sale_id INT,
     product_id INT,
     quantity INT,
     sale_date DATE
   );
   
   INSERT INTO product_sales VALUES
     (101, 1, 3, current_date()),
     (102, 2, 1, current_date()),
     (103, 4, 5, date_add(current_date(), -2)),
     (104, 6, 2, date_add(current_date(), -10));

2. Crie uma tabela de fatos que recalcula apenas os últimos 7 dias de ventas em cada refresh:

   SQL

   CREATE OR REFRESH STREAMING TABLE product_sales_recent
   SCHEDULE REFRESH EVERY 1 DAY
   FLOW REPLACE WHERE sale_date >= date_add(current_date(), -7) BY NAME
   SELECT sale_id, product_id, quantity, sale_date
   FROM product_sales;

   • FLOW REPLACE WHERE sale_date >= date_add(current_date(), -7): a cada refresh, exclui as linhas no destino que correspondem ao predicado, recalcula a query de origem para esse mesmo intervalo e insere os resultados. As linhas fora da janela de 7 dias permanecem intocadas.
   • SCHEDULE REFRESH EVERY 1 DAY: realiza o refresh da tabela no mesmo programar diário da tabela de dimensão.
   • BY NAME: obrigatório. Combina colunas pelo nome em vez da posição.

   A venda datada de 10 dias atrás cai fora do predicado, portanto o fluxo não a carrega. Apenas as ventas dos últimos 7 dias são mantidas.

3. query a tabela de fatos:

   SQL

   SELECT sale_id, product_id, quantity, sale_date
   FROM product_sales_recent
   ORDER BY sale_date DESC, sale_id;

   O resultado contém as três ventas dos últimos 7 dias. Os valores sale_date dependem do dia em que você faz a execução do tutorial:

   id_venda	product_id	quantidade	data_venda
   101	1	3	Hoje
   102	2	1	Hoje
   103	4	5	há 2 dias

   id_venda	product_id	quantidade	data_venda
   101	1	3	Hoje
   102	2	1	Hoje
   103	4	5	há 2 dias

   A venda 104 não está no resultado porque sua data está fora da janela de 7 dias.

O passo 7: Criar uma view materializada agregada

Agora que você tem uma tabela de dimensões que se mantém atualizada com as alterações da fonte, você pode adicionar uma camada de relatórios por cima.

Uma viewmaterializada armazena os resultados de consultas pré-computadas como uma tabela física. Ao contrário de uma view regular, que reexecuta a consulta cada vez que você lê dela, uma view materializada persiste os resultados e recalcula apenas as linhas afetadas por alterações upstream a cada refresh. Isso o torna ideal para dashboards e relatórios onde o desempenho das consultas é importante.

SQL

CREATE OR REPLACE MATERIALIZED VIEW products_by_category
SCHEDULE REFRESH EVERY 1 DAY
AS
SELECT
  category,
  COUNT(*) AS active_products
FROM products_history
WHERE __END_AT IS NULL
GROUP BY category;

SCHEDULE REFRESH EVERY 1 DAY significa que esta view faz refresh programar. Combinado com o mesmo programar na tabela de transmissão, você agora tem um pipeline de três estágios onde as alterações na tabela de origem ocorrem em cascata através da dimensão e para o agregado em cada ciclo de refresh. Você não precisa dar refresh manualmente na dimensão ou no agregado.

SQL

SELECT * FROM products_by_category ORDER BY active_products DESC;

o passo 8: Verificar a cascata de ponta a ponta

Para verificar a cascata completa do pipeline, faça uma alteração na tabela de origem e registre uma venda que chegou com atraso ontem:

SQL

UPDATE products SET warehouse = 'Seattle' WHERE product_id = 3;

INSERT INTO product_sales VALUES (105, 3, 4, date_add(current_date(), -1));

The Knife vai de Denver para Seattle. O UPDATE Trigger a cascata completa do pipeline, demonstrando como os três estágios funcionam juntos:

1. products registra o evento de alteração no feed de dados de alteração.
2. products_history Processa o evento e adiciona uma nova versão para a Faca.
3. products_by_category Recalcula apenas a linha de talheres afetada.

A tabela de fatos realiza o refresh de acordo com seu próprio programar. Para coletar a venda que chegou atrasada agora em vez de esperar pelo próximo refresh agendado, realize o refresh da tabela de fatos manualmente:

SQL

REFRESH STREAMING TABLE product_sales_recent;

O fluxo recalcula os últimos 7 dias a partir de product_sales, portanto, a venda 105 agora aparece em product_sales_recent.

Verificar:

SQL

SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
WHERE product_id = 3
ORDER BY __START_AT;

SELECT * FROM products_by_category ORDER BY active_products DESC;

SELECT sale_id, product_id, quantity, sale_date
FROM product_sales_recent
ORDER BY sale_date DESC, sale_id;

Limpar

Para limpar o recurso criado por este tutorial, use o seguinte SQL:

SQL

DROP MATERIALIZED VIEW IF EXISTS products_by_category;
DROP TABLE IF EXISTS product_sales_recent;
DROP TABLE IF EXISTS product_sales;
DROP TABLE IF EXISTS products_history;
DROP TABLE IF EXISTS products;

Recursos adicionais

• Tabelas de streaming
• Visualizações materializadas
• Cláusula AUTO CDC FLOW
• Fluxos REPLACE WHERE para tabelas de transmissão autônomas
• Quando usar REPLACE WHERE vs. visualizações materializadas
• APIs AUTO CDC : Simplifique a captura de dados de alterações (CDC) com pipeline
• Spark Declarative Pipelines no Lakeflow