Databricks Certified Data Engineer Professional

Projeto Python + bundle

meu_projeto/
├── databricks.yml          # bundle: recursos e targets
├── resources/              # jobs.yml, pipelines.yml (separados por recurso)
├── src/meu_projeto/        # pacote Python importável (transformações puras)
│   └── transformacoes.py
├── tests/                  # pytest
└── pyproject.toml          # gera o wheel

O bundle pode construir o wheel (artifacts) e anexá-lo às tarefas (libraries: - whl: ../dist/*.whl). Código de negócio fica no pacote, os notebooks só orquestram — isso torna tudo testável.

Dependências

• Serverless: dependências no environment da tarefa/notebook (environment_version + lista de pacotes), não em init scripts (serverless não tem init script).
• Pipelines: dependências na configuração do pipeline (environment) ou %pip install no topo do código-fonte.
• Conflito (ImportError, versão errada): fixe versões, evite instalar por cima do runtime, e confira se um wheel local também está no artefato do bundle.

UDFs: qual usar

Tipo	Quando	Observação
Função nativa (F.*)	Sempre que existir	Mais rápida: otimizada pelo Catalyst/Photon
SQL UDF / função do UC	Lógica reutilizável e governada	CREATE FUNCTION main.f.limpa(x STRING) RETURNS STRING RETURN trim(lower(x))
Pandas UDF (@pandas_udf)	Lógica em Python vetorizada	Usa Arrow: processa lotes, muito mais rápida que UDF Python linha a linha
UDF Python escalar	Último recurso	Serializa linha a linha

Streaming table x materialized view

• Streaming table: lê cada linha nova uma vez (append). Ideal para ingestão e baixa latência. Não "corrige" o passado.
• Materialized view: mantém o resultado completo de uma query (agregações, joins), refrescada incrementalmente quando possível. Reflete atualizações tardias em dias anteriores.

AUTO CDC (SCD 1 e 2)

from pyspark import pipelines as dp

dp.create_streaming_table("clientes_silver")

dp.create_auto_cdc_flow(
    target="clientes_silver",
    source="clientes_cdc_bronze",
    keys=["cliente_id"],
    sequence_by="ts_evento",                 # ordena eventos fora de ordem
    apply_as_deletes="operacao = 'DELETE'",
    except_column_list=["operacao", "ts_evento"],
    stored_as_scd_type=2,                    # 1 = sobrescreve; 2 = histórico com __START_AT/__END_AT
)

Em SQL: CREATE FLOW ... AS AUTO CDC INTO clientes_silver FROM STREAM(bronze) KEYS (cliente_id) SEQUENCE BY ts_evento STORED AS SCD TYPE 2.

Structured Streaming x pipelines declarativos

Pipelines declarativos cuidam de dependências, checkpoints, retries, expectations e infraestrutura. Structured Streaming "puro" dá controle total (ex.: foreachBatch com lógica arbitrária, sinks customizados). Pergunta fala em "menos operação/gestão" → pipeline declarativo.

Estado no streaming

(eventos
  .withWatermark("ts", "30 minutes")                  # limita o estado: descarta o que chegar >30 min atrasado
  .groupBy(F.window("ts", "10 minutes"), "loja")
  .count()
  .writeStream
  .outputMode("append")                               # append exige watermark em agregações
  .option("checkpointLocation", "/Volumes/main/chk/contagem")
  .toTable("main.gold.contagem_por_loja"))

• Output modes: append (só linhas finais), update (linhas alteradas), complete (tabela inteira; estado cresce sem limite).
• Checkpoint = offsets + estado → reinício sem duplicar (exactly-once com sink idempotente/transacional como Delta).
• foreachBatch: cada micro-batch vira um DataFrame batch — use para MERGE em Delta.

Testes

from pyspark.testing import assertDataFrameEqual, assertSchemaEqual

def adiciona_total(df):
    return df.withColumn("total", F.col("qtd") * F.col("preco"))

def test_total(spark):
    entrada = spark.createDataFrame([(2, 5.0)], "qtd INT, preco DOUBLE")
    esperado = spark.createDataFrame([(2, 5.0, 10.0)], "qtd INT, preco DOUBLE, total DOUBLE")
    assertDataFrameEqual(entrada.transform(adiciona_total), esperado)

DataFrame.transform encadeia funções puras — fáceis de testar uma a uma.

Compute

Serverless com performance mode (prioriza velocidade de start/execução) ou standard (mais barato, pode demorar mais). Notebook que precisa de muita RAM → tarefa com high-memory. Para ETL não idempotente, desligue retries automáticos.

Mensageria

(spark.readStream.format("kafka")
   .option("kafka.bootstrap.servers", "broker:9092")
   .option("subscribe", "pedidos")
   .option("startingOffsets", "earliest")
   .load()
   .select(F.col("value").cast("string").alias("payload"), "timestamp"))

Kinesis (format("kinesis")) e Pub/Sub (format("pubsub")) seguem a mesma ideia. Arquivos binários (PDF, imagem): format("binaryFile") ou Auto Loader com cloudFiles.format = binaryFile.

CDC de bancos com Lakeflow Connect

Conector gerenciado para SQL Server, MySQL e PostgreSQL: um gateway lê o log de mudanças do banco e um pipeline de ingestão aplica inserts/updates/deletes nas tabelas do UC. Zero código, incremental, governado. Alternativas (JDBC full load noturno, exportar CSV) perdem deletes ou custam caro.

Destino Iceberg

Tabelas gerenciadas do UC podem ser Iceberg (ou Delta com UniForm, legível por clientes Iceberg). Pipelines podem gravar nelas, inclusive com AUTO CDC.

OpenSharing (Delta Sharing) e Clean Rooms

• Databricks-to-Databricks (D2D): o destinatário tem workspace com UC; compartilha tabelas, views, volumes, modelos, notebooks. Sem cópia.
• Databricks-to-Open: destinatário fora da Databricks (pandas, Spark, Power BI…) usa um arquivo de credencial / token ou OIDC.
• Clean Rooms: duas ou mais partes analisam dados juntas sem expor os dados brutos uma à outra; só notebooks/saídas aprovados rodam.

CREATE SHARE vendas_parceiro;
ALTER SHARE vendas_parceiro ADD TABLE main.gold.vendas_agregadas;
CREATE RECIPIENT parceiro_x;                       -- open: gera link de ativação
GRANT SELECT ON SHARE vendas_parceiro TO RECIPIENT parceiro_x;

Lakehouse Federation

Consultar Postgres, MySQL, SQL Server, Snowflake, BigQuery, Redshift… sem copiar:

CREATE CONNECTION pg_erp TYPE postgresql
  OPTIONS (host 'erp.db', port '5432', user secret('scope','u'), password secret('scope','p'));
CREATE FOREIGN CATALOG erp USING CONNECTION pg_erp OPTIONS (database 'erp');
GRANT USE CATALOG, USE SCHEMA, SELECT ON CATALOG erp TO `analistas`;

A credencial fica na conexão (os usuários não a veem) e o acesso é controlado por GRANTs do UC, com linhagem e auditoria.

Janelas

SELECT *,
  ROW_NUMBER() OVER (PARTITION BY cliente_id ORDER BY ts DESC)        AS rn,
  SUM(valor)  OVER (PARTITION BY cliente_id ORDER BY ts
                    ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)  AS acumulado,
  LAG(valor)  OVER (PARTITION BY cliente_id ORDER BY ts)               AS valor_anterior
FROM main.silver.pedidos
QUALIFY rn = 1;   -- último pedido de cada cliente

VARIANT

CREATE TABLE main.bronze.eventos AS
SELECT parse_json(payload) AS v FROM main.raw.eventos_json;

SELECT v:usuario.id::BIGINT          AS usuario_id,   -- caminho com ':'
       variant_get(v, '$.valor', 'DOUBLE') AS valor,
       try_variant_get(v, '$.cupom', 'STRING') AS cupom  -- NULL se o tipo não bate
FROM main.bronze.eventos;

VARIANT guarda JSON num formato binário eficiente, sem schema fixo, e a extração é muito mais rápida que get_json_object em STRING.

AI functions

SELECT ticket_id,
       ai_classify(texto, ARRAY('cobrança', 'entrega', 'produto')) AS categoria,
       ai_query('databricks-meta-llama-3-3-70b-instruct',
                CONCAT('Resuma em uma frase: ', texto))           AS resumo
FROM main.silver.tickets;

ai_query chama qualquer endpoint de serving (fundação ou seu modelo) em batch, paralelizado.

Expectations: as três ações

Sintaxe	Efeito
EXPECT (cond)	Mantém a linha, registra métrica (warn)
... ON VIOLATION DROP ROW	Descarta a linha
... ON VIOLATION FAIL UPDATE	Falha o update do pipeline

Quarentena: inverta a regra numa segunda tabela.

regras = {"id_ok": "id IS NOT NULL", "valor_ok": "valor >= 0"}
cond = " AND ".join(f"({r})" for r in regras.values())

@dp.table
@dp.expect_all_or_drop(regras)
def pedidos_validos(): return spark.readStream.table("bronze_pedidos")

@dp.table
def pedidos_quarentena():
    return spark.readStream.table("bronze_pedidos").where(f"NOT ({cond})")

System tables que caem

Tabela	Para quê
system.billing.usage	DBUs por SKU, job, warehouse, usuário (usage_metadata)
system.billing.list_prices	Preço de tabela por SKU → custo = DBU × preço
system.compute.clusters, warehouses	Configuração e histórico de compute
system.access.audit	Quem fez o quê (logins, GRANTs, leituras)
system.lakeflow.jobs, job_run_timeline, job_task_run_timeline	Execuções, duração, status
system.query.history	Queries de SQL warehouses e serverless

SELECT u.usage_metadata.job_id,
       SUM(u.usage_quantity * p.pricing.default) AS custo_usd
FROM system.billing.usage u
JOIN system.billing.list_prices p
  ON u.sku_name = p.sku_name
 AND u.usage_start_time BETWEEN p.price_start_time AND COALESCE(p.price_end_time, now())
WHERE u.usage_date >= date_trunc('month', current_date() - INTERVAL 1 MONTH)
GROUP BY 1 ORDER BY 2 DESC LIMIT 10;

API, CLI e SDK

databricks jobs list-runs --job-id 123 --limit 20
databricks pipelines list-pipeline-events <pipeline-id>

from databricks.sdk import WorkspaceClient
w = WorkspaceClient()
for run in w.jobs.list_runs(job_id=123, limit=20):
    print(run.run_id, run.state.result_state, run.run_duration)

Event log do pipeline

SELECT timestamp, details:flow_progress.data_quality.expectations
FROM event_log(TABLE(main.silver.pedidos))
WHERE event_type = 'flow_progress';

Traz progresso dos flows, linhas processadas, resultado das expectations, erros e linhagem.

Alerts

Um alert roda uma query num agendamento e notifica (e-mail, Slack, webhook) quando a condição bate — ex.: custo diário acima de X, % de linhas rejeitadas, queries lentas. Também pode servir de tarefa num job para ramificar o fluxo.

Menos manutenção

• Tabela gerenciada do UC + Predictive Optimization: OPTIMIZE, VACUUM e ANALYZE automáticos.
• Liquid Clustering: layout incremental, sem escolher partições; CLUSTER BY AUTO deixa a Databricks escolher (e trocar) as chaves pelo histórico de queries.

Qual técnica para qual padrão

Padrão	Técnica
Muitos MERGE/UPDATE/DELETE pequenos	Deletion vectors (marca linhas removidas sem reescrever o arquivo)
Filtros frequentes em 1–4 colunas, alta cardinalidade	Liquid Clustering nessas colunas
Padrão de consulta muda ou é desconhecido	CLUSTER BY AUTO
Tabela pequena (< ~1 TB)	Geralmente nada além de clustering/auto — não particione

Liquid x partição/ZORDER

Partição só ajuda com coluna de baixa cardinalidade e tabelas muito grandes; partição de alta cardinalidade gera milhões de arquivos pequenos. ZORDER não é incremental e não muda de chave sem reescrever. Liquid é incremental, aceita mudar chaves e funciona com alta cardinalidade.

Cache

O disk cache (antigo Delta cache) guarda cópias locais (SSD) dos arquivos Parquet lidos → leituras repetidas dos mesmos dados ficam rápidas. É automático em instâncias com SSD e no SQL warehouse. Diferente do df.cache() do Spark (memória do executor, por sessão).

Change Data Feed

ALTER TABLE main.silver.clientes SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

SELECT * FROM table_changes('main.silver.clientes', 120);  -- desde a versão 120
-- colunas extras: _change_type (insert, update_preimage, update_postimage, delete), _commit_version

spark.readStream.option("readChangeFeed", "true").table("main.silver.clientes")

O downstream processa só o que mudou, inclusive updates e deletes.

Query profile

Mostra por operador: linhas, tempo, arquivos lidos x podados (data skipping), tipo de join (broadcast hash, sort-merge, shuffle hash) e bytes de shuffle. Sinais:

• "files read" quase igual a "files total" com filtro seletivo → layout ruim → Liquid Clustering na coluna do filtro;
• sort-merge com um lado pequeno → broadcast;
• shuffle enorme → filtrar antes, pré-agregar, cluster nas chaves de join.

Menor privilégio

Conceda a grupos, no menor escopo que resolve, apenas o privilégio necessário (SELECT em vez de ALL PRIVILEGES; MODIFY só para quem escreve). Jobs rodam como service principal. Recursos do workspace (jobs, warehouses, notebooks) têm ACLs próprias (CAN VIEW, CAN RUN, CAN MANAGE).

ABAC com governed tags

-- 1) marcar colunas (ou deixar a classificação automática sugerir)
ALTER TABLE main.crm.clientes ALTER COLUMN cpf SET TAGS ('pii' = 'cpf');

-- 2) uma política vale para o catálogo todo, inclusive tabelas futuras
--    (sintaxe resumida: confira a forma exata na documentação de ABAC)
CREATE POLICY mascarar_pii ON CATALOG main
  COLUMN MASK main.seg.mascara_pii
  TO `account users` EXCEPT `compliance`
  FOR TABLES MATCH COLUMNS has_tag('pii') AS col ON COLUMN col;

Técnicas de anonimização

Técnica	Exemplo	Reversível?
Hash (com salt)	sha2(concat(cpf, :salt), 256)	Não; mas permite join entre tabelas
Tokenização	troca o valor por um token; tabela de mapeamento protegida	Sim, para quem acessa o cofre
Supressão	'***' ou NULL	Não
Generalização	idade 37 → faixa "30–39"; CEP → 3 primeiros dígitos	Não

Pipeline com PII

Detecte (classificação de dados do UC / ai_mask / regex), aplique máscara ou hash na silver e mantenha o dado bruto num schema restrito. Funciona igual em batch e streaming porque a máscara fica na tabela, não no código de cada consumidor.

Direito ao esquecimento

DELETE FROM main.silver.clientes WHERE cliente_id = 42;   -- com deletion vectors: rápido
REORG TABLE main.silver.clientes APPLY (PURGE);            -- reescreve os arquivos sem as linhas
VACUUM main.silver.clientes;                               -- remove arquivos antigos após a retenção

Sem VACUUM, o dado ainda existe em arquivos antigos (time travel). Lembre-se de propagar o delete para tabelas derivadas (CDF ajuda) e de checar retenção de logs.

Tags e comentários

COMMENT ON TABLE main.gold.vendas IS 'Vendas diárias consolidadas por loja (fonte: ERP).';
ALTER TABLE main.gold.vendas ALTER COLUMN receita COMMENT 'Receita líquida em BRL';
ALTER TABLE main.gold.vendas SET TAGS ('dominio' = 'comercial', 'certificado' = 'sim');

Comentários e tags aparecem no Catalog Explorer, na busca e são usados por Genie/assistentes para entender os dados. Governed tags têm valores controlados e podem disparar políticas ABAC.

Herança

Privilégio concedido em catálogo vale para todos os schemas e objetos dentro dele — inclusive os criados depois. O mesmo vale de schema para tabelas. Não existe "DENY" no UC: a restrição se faz não concedendo, ou com filtros/máscaras. O owner de um objeto pode concedê-lo; owner do catálogo controla tudo abaixo.

Onde olhar

Pergunta	Fonte
Qual stage/task está lenta? skew? spill?	Spark UI (Stages, SQL/DataFrame)
Erro de biblioteca, init script, OOM do driver	Logs do driver/cluster e event log do cluster
Tendência de duração/custo, falhas em massa	System tables (lakeflow, billing, query.history)
Por que esta query SQL é lenta?	Query profile

Repair run

Quando 2 de 10 tarefas falham, repair run reexecuta só as que falharam e as dependentes, reaproveitando o que já deu certo — e permite sobrescrever parâmetros (ex.: corrigir uma data). Mais rápido e barato que rodar o job inteiro.

databricks jobs repair-run 987654 --rerun-tasks transformar --job-parameters '{"data":"2026-10-18"}'

Bundles em CI/CD

databricks bundle validate → deploy -t staging → testes de integração → deploy -t prod com service principal. Targets definem host, variáveis, permissões e mode.

Git folders

Para notebooks e código no workspace: branch por feature, PR no provedor Git, e em produção um Git folder fixado na branch principal atualizado pela API/CLI no pipeline de CI (databricks repos update <id> --branch main) — ou, melhor, bundles.

Layout

• Partição só pela coluna que bate com o grão de carga/expurgo e tem baixa cardinalidade (ex.: data em tabelas de muitos TB).
• Clustering pelas colunas de filtro/join mais usadas.
• Compactação (OPTIMIZE, ou automática) mantém arquivos de tamanho saudável (~100 MB–1 GB) e evita o problema dos arquivos pequenos.

Dimensional + MVs + Metric Views

Fato (eventos, medidas) + dimensões (cliente, produto, tempo). Por cima:

• Materialized views pré-calculam agregações pesadas que os dashboards leem.
• Metric views definem a métrica uma vez, governada no UC:

version: 1.1
source: main.gold.fato_vendas
dimensions:
  - name: loja
    expr: loja_id
  - name: mes
    expr: date_trunc('MONTH', data_venda)
measures:
  - name: receita_liquida
    expr: SUM(valor_bruto - descontos - devolucoes)

SELECT mes, MEASURE(receita_liquida) FROM main.gold.mv_vendas GROUP BY mes;

Resolve "cada dashboard calcula a receita de um jeito".