ETL em Databricks SQL

Ao lidar com grandes quantidades de dados, é necessário um pipeline que possa processar apenas os registos novos e alterados, em vez de reprocessar todo o conjunto de dados. Isto chama-se ETL incremental. No Databricks SQL, pode construir pipelines ETL incrementais usando tabelas de streaming e vistas materializadas, sem escrever código procedural ou agendar atualizações manuais.

Este tutorial guia-o por um padrão comum: acompanhar as alterações do produto ao longo do tempo. Cria-se uma tabela de origem, capta eventos de alteração, constrói uma tabela de dimensões que preserva o histórico completo de cada produto, mantém uma tabela de factos que sobrescrive progressivamente registos recentes e adiciona uma camada agregada de relatórios por cima. Usas AUTO CDC para a tabela de dimensões e um fluxo REPLACE WHERE para a tabela de factos.

A principal funcionalidade deste tutorial é AUTO CDC. Num armazém tradicional, escreveria instruções complexas MERGE INTO para conciliar a inserção, atualização e eliminação de eventos numa tabela alvo. Esta abordagem é propensa a erros, especialmente quando os acontecimentos surgem fora de ordem. AUTO CDC Trata disto por ti. Declaras a chave de negócio, a coluna de sequenciação e se queres SCD Tipo 1 (apenas o valor mais recente) ou SCD Tipo 2 (histórico completo), e o Azure Databricks aplica automaticamente a lógica de fusão correta. Para uma visão geral do CDC, veja As APIs AUTO CDC – Simplifique a captura de dados de alterações com pipelines.

Ao final deste tutorial, você terá:

  1. Criei uma tabela de origem que acompanha as alterações com o Change Data Feed.
  2. Inspecionei os dados brutos das alterações para compreender o fluxo de eventos do CDC.
  3. Usado AUTO CDC para construir uma tabela de dimensões SCD Tipo 2 a partir desses eventos.
  4. Processa os eventos de eliminação incrementalmente através do pipeline.
  5. Criou-se uma tabela de factos que sobrescreve incrementalmente as linhas recentes com um fluxo REPLACE WHERE.
  6. Criei uma vista materializada que mantém um relatório agregado de forma incremental.
  7. Configurado SCHEDULE REFRESH EVERY 1 DAY para que as alterações se propaguem automaticamente ao longo da pipeline.

Requisitos

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

Passo 1: Configure o seu catálogo e esquema

Abre o editor SQL do Databricks e define o teu catálogo e esquema de trabalho. É necessário ter permissão para USE o catálogo e o esquema que selecionar:

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

Passo 2: Criar uma tabela de origem e carregar os dados

Crie uma tabela products com o feed de dados de alterações no Azure Databricks (CDF) ativado. O CDF é uma funcionalidade Delta Lake que regista cada inserção, atualização e eliminação como um registo de alterações consultável. Isto é 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 num log externo. Aqui utiliza-se o CDF para gerar os eventos de alteração que o pipeline a jusante irá consumir.

  1. Crie a tabela e carregue os registos 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 a montante, incluindo novos produtos, uma mudança 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;

Passo 3: Consultar o feed de dados de alterações

Antes de construir o pipeline a jusante, ajuda olhar para os eventos brutos de alteração para perceber o que AUTO CDC irá processar. A table_changes() função lê o registo do CDF e devolve todas as operações capturadas juntamente com as 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).

Note que uma única alteração lógica (por exemplo, mover a Colher para outro armazém) produz múltiplos eventos: uma pré-imagem e uma pós-imagem. Num armazém tradicional, escreveria uma MERGE instrução para reconciliar todos estes eventos numa tabela de alvo, tratando inserções, atualizações e eliminações com lógica separada, garantindo que os eventos são aplicados na ordem correta. Esta é exatamente a complexidade que AUTO CDC elimina no passo seguinte.

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

Uma tabela de streaming processa os dados de forma incremental. Em cada atualização, só lê as novas linhas desde a última execução, por isso não precisa de reprocessar o conjunto de dados completo. Isto torna-o ideal para fontes de grande volume ou frequentemente mutáveis.

AUTO CDC adiciona processamento de captura de dados de alterações por cima de uma tabela de streaming. Em vez de escrever uma instrução MERGE INTO que gere manualmente inserções, atualizações e eliminações, declara a chave de negócio e a coluna de sequenciação e deixa-Azure Databricks aplicar a lógica correta. AUTO CDC também processa automaticamente eventos fora de ordem, que é um problema comum quando se utiliza MERGE INTO para lidar com eventos que chegam de sistemas distribuídos ou cargas em lote com carimbos de tempo 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 __START_AT e __END_AT carimbos temporais. A NULL em __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: refresca a mesa num horário diário.
  • FLOW AUTO CDC: declara isto como um fluxo CDC. O Azure Databricks aplica automaticamente as semânticas de inserção, atualização e eliminação.
  • WHERE _change_type != 'update_preimage': salta preimages. A postimage já tem a linha atualizada.
  • KEYS (product_id): a chave de negócios. Eventos com a mesma chave são fundidos em linhas versionadas.
  • APPLY AS DELETE WHEN _change_type = 'delete': fecha a versão atual quando chega um evento de eliminação. Isto permite-lhe definir a condição que identifica um evento de eliminação.
  • SEQUENCE BY _commit_timestamp: estabelece a ordenação dos eventos. Lida corretamente com chegadas fora de ordem.
  • STORED AS SCD TYPE 2: conserva toda a história. AUTO CDC suporta tanto SCD Tipo 1 como SCD Tipo 2.

Consultar 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 set) e Los Angeles (atual, __END_AT = NULL).
  • Garfo: duas versões. Categoria de talheres (fechada) e categoria de jantar (atual).
  • Guardanapo e Porta-copos: uma versão cada (recém-inseridos, __END_AT = NULL).
  • Todos os outros produtos: uma versão para cada (__END_AT = NULL).

Passo 5: Eliminações de processos através do pipeline

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

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

Estes eventos de eliminação são registados no registo 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 foram aplicadas as eliminações:

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

Bowl e Glass estão agora encerrados com o conjunto __END_AT, sendo marcados como descontinuados. Todos os outros produtos atuais mantêm-se inalterados. A tabela de streaming só processava os novos eventos de eliminação sem reprocessar as inserções e atualizações da atualização anterior.

Passo 6: Constrói uma tabela de factos 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. As tabelas de factos têm necessidades diferentes. Uma tabela de factos regista eventos, como vendas, e normalmente só se reprocessa uma janela recente, por exemplo, para corrigir registos de chegada tardia. Para este padrão, use um fluxo REPLACE WHEREincremental, que sobrescrive apenas as linhas que correspondem a um predicado e deixa todas as outras linhas intocadas. Para uma comparação completa desta construção com vistas materializadas, veja Quando usar vistas REPLACE WHERE vs. vistas materializadas.

  1. Crie uma product_sales tabela e carregue 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 factos 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): em cada atualização, elimina as linhas no destino que correspondem ao predicado, recalcula a consulta de origem para esse mesmo intervalo e insere os resultados. As filas fora da janela de 7 dias ficam intocadas.
    • SCHEDULE REFRESH EVERY 1 DAY: atualiza a tabela segundo o mesmo agendamento diário da tabela de dimensões.
    • BY NAME: obrigatório. Faz a correspondência das colunas pelo nome em vez de por posição.

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

  3. Consulta a tabela de factos:

    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 ocorridas nos últimos 7 dias. Os sale_date valores dependem do dia em que executa o tutorial:

    sale_id product_id quantidade data_venda
    101 1 3 Hoje
    102 2 1 Hoje
    103 4 5 Há 2 dias

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

Passo 7: Criar uma vista materializada agregada

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

Uma visualização materializada armazena resultados de consultas pré-calculadas como uma tabela física. Ao contrário de uma vista normal, que reexecuta a consulta cada vez que é lida, uma vista materializada preserva os resultados e recalcula apenas as linhas afetadas por alterações a montante em cada atualização. Isto torna-o bem adequado para painéis e relatórios onde o desempenho das consultas é relevante.

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 vista se atualiza num horário diário. Combinado com o mesmo cronograma na tabela de streaming, tens agora um pipeline de três etapas onde as alterações à tabela de origem passam pela dimensão e entram no agregado em cada ciclo de atualização. Não precisas de atualizar manualmente a dimensão ou o agregado.

SELECT * FROM products_by_category ORDER BY active_products DESC;

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 registe uma venda com chegada tardia 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));

A Faca muda-se de Denver para Seattle. O UPDATE desencadeia toda a cascata do pipeline, demonstrando como os três estágios funcionam em conjunto:

  1. products regista o evento de alteração no change data feed.
  2. products_history processa o evento e adiciona uma nova versão para a faca.
  3. products_by_category recalcula apenas a fila de talheres afetada.

A tabela de factos atualiza-se de acordo com o seu próprio agendamento. Para recolher a venda recebida tardiamente agora, em vez de esperar pela próxima atualização agendada, atualize manualmente a tabela de factos:

REFRESH STREAMING TABLE product_sales_recent;

O fluxo recalcula os últimos 7 dias a partir de product_sales, pelo que a venda 105 aparece agora 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 a