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 conjunto de dados. Isso é chamado de ETL incremental. No Databricks SQL, você pode criar pipelines ETL incrementais usando tabelas de streaming e exibições materializadas, sem escrever código de procedimento ou agendar atualizações manuais.

Este tutorial orienta você por um padrão comum: acompanhar as alterações do produto ao longo do tempo. Você cria uma tabela de origem, captura eventos de alteração, cria uma tabela de dimensão que preserva o histórico completo de cada produto, mantém uma tabela de fatos que sobrescreve, de forma incremental, registros recentes e adiciona uma camada de relatórios agregados por cima. Você usa AUTO CDC para a tabela de dimensões e um fluxo REPLACE WHERE para a tabela de fatos.

O principal recurso neste tutorial é AUTO CDC. Em um armazém tradicional, você escreveria instruções complexas MERGE INTO para reconciliar 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 chave de negócios, a coluna de sequenciamento e se deseja scd tipo 1 (somente valor mais recente) ou SCD Tipo 2 (histórico completo) e Azure Databricks aplica a lógica de mesclagem correta automaticamente. Para obter uma visão geral do CDC, consulte As APIs AUTO CDC: simplificar a captura de dados de alteração com pipelines.

Ao final deste tutorial, você terá:

  1. Criada uma tabela de origem que acompanha as alterações com o Feed de Dados de Alterações.
  2. Inspecionados os dados brutos de alteração para entender o fluxo de eventos CDC.
  3. Foi usado AUTO CDC para criar uma tabela de dimensões SCD Tipo 2 a partir desses eventos.
  4. Eventos de exclusão processados incrementalmente por meio do pipeline.
  5. Construiu-se uma tabela de fatos que sobrescreve incrementalmente as linhas recentes com um fluxo REPLACE WHERE.
  6. Criou uma visão materializada que mantém de forma incremental um relatório agregado.
  7. Configurado SCHEDULE REFRESH EVERY 1 DAY para que as alterações sejam propagadas automaticamente por meio do pipeline.

Requisitos

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

Etapa 1: Configurar seu catálogo e esquema

Abra o editor de SQL do Databricks e defina seu catálogo e esquema de trabalho. Você deve ter permissão para USE o catálogo e o esquema selecionado:

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

Etapa 2: Criar uma tabela de origem e carregar dados

Crie uma tabela products com Usar o feed de dados de alterações no Azure Databricks (CDF) ativado. CDF é um recurso do Delta Lake que registra cada inserção, atualização e exclusão como um log de alterações que pode ser consultado. Isso é semelhante a um fluxo CDC de um sistema de origem transacional, exceto que as alterações são capturadas diretamente na tabela Delta, em vez de em um registro externo. Você utiliza o CDF aqui para gerar os eventos de alteração que serão consumidos pelo pipeline subsequente.

  1. Crie a tabela e carregue os registros iniciais:

    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 upstream, incluindo novos produtos, uma movimentação de armazém e uma reatribuição de categoria:

    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;

Etapa 3: Consultar o feed de dados de alteração

Antes de criar o pipeline downstream, é útil examinar os eventos de alteração brutos para que você possa entender o que AUTO CDC vai processar. A table_changes() função lê o log do CDF e retorna todas as operações capturadas junto com colunas de metadados:

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

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

Observe que uma única alteração lógica (por exemplo, mover o Spoon para um warehouse diferente) produz vários eventos: uma pré-imagem e uma postimagem. Em um armazém tradicional, você escreveria uma MERGE instrução para reconciliar todos esses eventos em uma tabela de destino, manipulando inserções, atualizações e exclusões com lógica separada e certificando-se de que os eventos sejam aplicados na ordem correta. Essa é exatamente a complexidade que AUTO CDC elimina na próxima etapa.

Etapa 4: Criar uma dimensão SCD Tipo 2 com AUTO CDC

Uma tabela de streaming processa dados incrementalmente. Em cada atualização, ela lê apenas as novas linhas desde a última execução, portanto, ela não precisa reprocessar o conjunto de dados completo. Isso o torna adequado para fontes de alto volume ou de alteração frequente.

AUTO CDC inclui o processamento de captura de dados de alteração na parte superior de uma tabela de streaming. Em vez de escrever uma instrução MERGE INTO que manipula manualmente inserções, atualizações e exclusões, você declara a chave de negócios e a coluna de sequenciamento e permite que Azure Databricks aplique a lógica correta. AUTO CDC também manipula 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 em lote com carimbos de data/hora sobrepostos.

A instrução a seguir cria uma tabela SCD Tipo 2 que preserva o histórico de versão completo de cada produto. Cada versão obtém __START_AT e __END_AT carimbos de data/hora. Um NULL in __END_AT marca a versão atual.

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 DAY: atualiza a tabela diariamente.
  • FLOW AUTO CDC: declara isso como um fluxo CDC. Azure Databricks aplica a inserção, atualização e exclusão de semântica automaticamente.
  • WHERE _change_type != 'update_preimage': ignora preimages. A postimage já tem a linha atualizada.
  • KEYS (product_id): a chave de negócio. Eventos com a mesma chave são mesclados em linhas versionadas.
  • APPLY AS DELETE WHEN _change_type = 'delete': fecha 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 ordenação de eventos. Lida com as chegadas fora de ordem corretamente.
  • STORED AS SCD TYPE 2: mantém o histórico completo. AUTO CDC dá suporte ao SCD Tipo 1 e SCD Tipo 2.

Consulte a tabela de dimensões:

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, __END_AT definido) e Los Angeles (atual, __END_AT = NULL).
  • Bifurcação: duas versões. Categoria de talher (fechada) e Categoria de jantar (atual).
  • Guardanapo e Porta-Copo: uma versão cada (recém-inseridos, __END_AT = NULL).
  • Todos os outros produtos: uma versão cada (__END_AT = NULL).

Etapa 5: Processar exclusões por meio do pipeline

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

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

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

REFRESH STREAMING TABLE products_history;

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

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

Tigela e Copo estão agora encerrados com o conjunto __END_AT, marcando-os como descontinuados. Todos os outros produtos atuais permanecem inalterados. A tabela de streaming só processou os novos eventos de exclusão sem reprocessar as inserções e atualizações da atualização anterior.

Passo 6: Construa uma tabela de fatos com um fluxo REPLACE WHERE

A tabela de dimensões nos passos anteriores usa AUTO CDC porque acompanha o estado atual e histórico de cada produto. Tabelas fato têm necessidades diferentes. Uma tabela de fatos registra eventos, como vendas, e normalmente você reprocessa apenas uma janela recente, por exemplo, para corrigir registros que chegam atrasados. Para esse 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 uma comparação completa dessa construção com visualizações materializadas, consulte Quando usar SUBSTITUIR WHERE vs. visualizações materializadas.

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

    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 recalcule apenas os últimos 7 dias de vendas em cada atualização:

    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 atualização, exclui as linhas no destino que correspondem ao predicado, recalcula a consulta de origem para o mesmo intervalo e insere os resultados. Linhas fora da janela de 7 dias permanecem intocadas.
    • SCHEDULE REFRESH EVERY 1 DAY: recarrega a tabela no mesmo cronograma diário da tabela de dimensões.
    • BY NAME: obrigatório. Combina colunas pelo nome em vez de por posição.

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

  3. Consulte a tabela de fatos:

    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 vendas dos últimos 7 dias. Os sale_date valores dependem do dia em que você conduz o tutorial:

    sale_id product_id quantidade data_de_venda
    101 1 3 Hoje
    102 2 1 Hoje
    103 4 5 2 dias atrás

    A Venda 104 não está no resultado porque sua data está fora do prazo de 7 dias.

Etapa 7: Criar uma visualização materializada agregada

Agora que você tem uma tabela de dimensões que permanece atual com as alterações de origem, você pode adicionar uma camada de relatório na parte superior.

Uma exibição materializada armazena os resultados da consulta pré-computada como uma tabela física. Ao contrário de uma exibição regular, que executa novamente a consulta sempre que você lê a partir dela, uma exibição materializada mantém os resultados e apenas recalcula as linhas afetadas por alterações upstream em cada atualização. Isso o torna adequado para dashboards e relatórios em que o desempenho da consulta importa.

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 essa exibição é atualizada em um agendamento diário. Combinado com o mesmo agendamento na tabela de streaming, agora você tem um pipeline de três estágios em que as alterações na tabela original propagam-se em cascata, atingindo a dimensão e a agregação em cada ciclo de atualização. Você não precisa atualizar manualmente a dimensão ou o agregado.

SELECT * FROM products_by_category ORDER BY active_products DESC;

Passo 8: Verifique 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 atrasada referente a ontem:

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

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

The Knife se muda de Denver para Seattle. UPDATE aciona toda a cascata 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ções.
  2. products_history processa o evento e adiciona uma nova versão para Faca.
  3. products_by_category recomputa apenas a linha Cutlery afetada.

A tabela de fatos é atualizada de acordo com seu próprio cronograma. Para incluir uma venda que chegou com atraso, em vez de aguardar a próxima atualização programada, atualize a tabela de fatos manualmente:

REFRESH STREAMING TABLE product_sales_recent;

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

Verificar:

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;

Limpeza

Para limpar os recursos criados por este tutorial, use o seguinte 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

  • Última atualização em