Databricks - Delta Lake

Delta Lake é o table format padrão do Databricks. Toda tabela criada no Databricks é Delta por padrão. Adiciona transações ACID, time travel, schema enforcement e operações de manutenção sobre arquivos Parquet no object storage.

Ver conceitos gerais em formatos-de-arquivo-datalake. Esta nota foca nas operações e otimizações específicas do Databricks.

Estrutura física de uma tabela Delta

s3://lake/silver/pedidos/
├── _delta_log/               ← transaction log (JSON + checkpoint Parquet)
│   ├── 00000000000000000000.json
│   ├── 00000000000000000001.json
│   └── 00000000000000000010.checkpoint.parquet
├── part-00000-abc123.parquet
├── part-00001-def456.parquet
└── part-00002-ghi789.parquet

O _delta_log é o que torna uma pasta de Parquets em uma tabela Delta, contendo o histórico de cada operação.

Operações DML

MERGE (Upsert)

from delta.tables import DeltaTable
 
delta_tabela = DeltaTable.forName(spark, "silver.pedidos")
 
delta_tabela.alias("alvo").merge(
    novos_dados.alias("origem"),
    "alvo.pedido_id = origem.pedido_id"
).whenMatchedUpdateAll() \
 .whenNotMatchedInsertAll() \
 .execute()

-- SQL equivalente
MERGE INTO silver.pedidos AS alvo
USING novos_pedidos AS origem
ON alvo.pedido_id = origem.pedido_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

MERGE é a base do padrão SCD (Slowly Changing Dimensions) e de pipelines CDC.

DELETE e UPDATE

DELETE FROM silver.pedidos WHERE status = 'cancelado' AND data_pedido < '2024-01-01';
 
UPDATE silver.pedidos
SET status = 'processado', updated_at = current_timestamp()
WHERE status = 'pendente' AND data_pedido < current_date();

Time Travel

-- Por versão
SELECT * FROM silver.pedidos VERSION AS OF 5;
 
-- Por timestamp
SELECT * FROM silver.pedidos TIMESTAMP AS OF '2026-01-15 08:00:00';
 
-- Ver histórico
DESCRIBE HISTORY silver.pedidos;

# Python
df = spark.read.format("delta").option("versionAsOf", 5).table("silver.pedidos")
df = spark.read.format("delta").option("timestampAsOf", "2026-01-15").table("silver.pedidos")

Time travel funciona enquanto os arquivos antigos não forem removidos pelo VACUUM.

OPTIMIZE e Z-ORDER

Com o tempo, tabelas Delta acumulam muitos arquivos pequenos (small files problem). OPTIMIZE os consolida.

-- Compactar arquivos pequenos
OPTIMIZE silver.pedidos;
 
-- Z-ORDER: co-localiza dados por coluna, acelerando queries com filtro nessa coluna
OPTIMIZE silver.pedidos ZORDER BY (cliente_id, data_pedido);

Z-ORDER reordena fisicamente os dados para que rows com os mesmos valores de cliente_id e data_pedido fiquem no mesmo arquivo, permitindo ao Spark pular arquivos inteiros que não contêm o valor filtrado (data skipping).

Liquid Clustering (recomendado para novas tabelas)

Liquid Clustering substitui o particionamento estático + Z-ORDER. Define colunas de clustering sem precisar reescrever a tabela.

-- Criar tabela com Liquid Clustering
CREATE TABLE silver.pedidos (
  pedido_id BIGINT,
  cliente_id BIGINT,
  data_pedido DATE,
  valor_total DECIMAL(10,2)
)
CLUSTER BY (cliente_id, data_pedido);
 
-- Aplicar clustering incremental (não reescreve tudo)
OPTIMIZE silver.pedidos;

Vantagens sobre particionamento + Z-ORDER:

Sem necessidade de escolher cardinalidade de partição no momento da criação
Rebalanceia incrementalmente, sem reescrever a tabela inteira
Recomendado para tabelas com > 1 TB ou queries com filtros múltiplos

VACUUM

Remove arquivos Parquet que não fazem mais parte da tabela (referências deletadas, versões antigas).

-- Remover arquivos mais antigos que 7 dias (padrão)
VACUUM silver.pedidos;
 
-- Remover arquivos mais antigos que 24 horas (atenção: apaga histórico de time travel)
VACUUM silver.pedidos RETAIN 24 HOURS;

Cuidado: após VACUUM, time travel para versões antigas desses dados não é mais possível.

Auto Optimize (configuração de tabela)

-- Habilitar compactação automática de arquivos pequenos após cada write
ALTER TABLE silver.pedidos
SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact'   = 'true'
);

Ou globalmente no cluster:

spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled",   "true")

Change Data Feed (CDF)

Permite consumir apenas as mudanças (inserts, updates, deletes) desde uma versão anterior, servindo de base para pipelines incrementais.

-- Habilitar na tabela
ALTER TABLE silver.pedidos SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true');
 
-- Ler mudanças a partir da versão 5
SELECT * FROM table_changes('silver.pedidos', 5);

df_changes = (
    spark.read.format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", 5)
    .table("silver.pedidos")
)
# _change_type: insert | update_preimage | update_postimage | delete

Ver também: databricks | databricks-unity-catalog | databricks-lakeflow-pipelines | formatos-de-arquivo-datalake | arquitetura-medalhao

Second Brain

Explorer