Databricks Certified Data Engineer Associate

A arquitetura em uma frase

A Databricks separa control plane (a interface, o agendador de jobs, o metastore do Unity Catalog — gerenciados pela Databricks) do compute plane (onde o Spark de fato roda: serverless, na conta da Databricks, ou clássico, na sua nuvem). Os dados ficam no object storage (S3, ADLS, GCS) em formato aberto.

Delta Lake: por que ele resolve tanta coisa

Delta é Parquet + um log de transações (_delta_log). Isso dá:

• ACID: escritas concorrentes não corrompem a tabela;
• Time travel: SELECT * FROM vendas VERSION AS OF 12 ou TIMESTAMP AS OF '2026-05-01';
• Schema enforcement (rejeita colunas inesperadas) e schema evolution (mergeSchema);
• MERGE, UPDATE, DELETE em data lake.

DESCRIBE HISTORY main.vendas.pedidos;           -- quem mudou o quê e quando
RESTORE TABLE main.vendas.pedidos TO VERSION AS OF 41;  -- desfaz uma carga ruim

Unity Catalog em uma frase

Governança única para dados e IA: namespace de 3 níveis catalogo.schema.objeto, permissões com GRANT, linhagem automática e auditoria.

Qual compute escolher

Compute	Quando usar	Custo
Serverless (notebooks/jobs)	Padrão recomendado: sobe em segundos, sem tuning	Paga só o uso, DBU serverless
Job compute (clássico)	Jobs agendados que precisam de configuração específica (init scripts, instâncias)	Mais barato que all-purpose; morre ao fim do job
All-purpose	Desenvolvimento interativo compartilhado	Mais caro; fica ligado (use auto-termination)
SQL warehouse (serverless/pro/classic)	BI, dashboards, SQL ad hoc com muitos usuários	Escala por concorrência

Regra de prova: ETL agendado → job compute ou serverless jobs; muitos analistas fazendo SQL → SQL warehouse; exploração em notebook → serverless ou all-purpose com auto-termination.

O mapa das opções de ingestão

Ferramenta	Fonte típica	Incremental?	Quando escolher
Upload pela UI	CSV/JSON pequeno do seu computador	Não	Carga pontual, protótipo
COPY INTO	Arquivos no object storage	Sim (idempotente: lembra o que já carregou)	Milhares de arquivos, carga batch simples em SQL
Auto Loader (cloudFiles)	Arquivos no object storage	Sim (checkpoint)	Milhões de arquivos, chegada contínua, schema que muda
Lakeflow Connect – conectores gerenciados	SaaS (Salesforce, Workday…) e bancos (SQL Server, Postgres, MySQL)	Sim, com CDC	Quer zero código e governança no UC
Lakeflow Connect – conectores padrão	Kafka, Kinesis, object storage, SFTP	Sim	Você escreve o pipeline, mas com conector pronto
JDBC/REST em notebook	API ou banco sem conector	Você implementa	Último recurso; agende com Lakeflow Jobs

COPY INTO

CREATE TABLE IF NOT EXISTS main.bronze.pedidos;

COPY INTO main.bronze.pedidos
FROM 's3://empresa-raw/pedidos/'
FILEFORMAT = JSON
FORMAT_OPTIONS ('inferSchema' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');

Rodar de novo não duplica: arquivos já carregados são ignorados. Para recarregar tudo, COPY_OPTIONS ('force' = 'true').

Auto Loader

(spark.readStream.format("cloudFiles")
   .option("cloudFiles.format", "json")
   .option("cloudFiles.schemaLocation", "/Volumes/main/bronze/_schemas/pedidos")
   .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
   .load("s3://empresa-raw/pedidos/")
 .writeStream
   .option("checkpointLocation", "/Volumes/main/bronze/_checkpoints/pedidos")
   .trigger(availableNow=True)          # roda como batch incremental e para
   .toTable("main.bronze.pedidos"))

• schemaLocation guarda o schema inferido e sua evolução.
• schemaEvolutionMode: addNewColumns (padrão: para o stream, adiciona a coluna, reinicie), rescue (joga o desconhecido em _rescued_data), failOnNewColumns, none.
• Directory listing (padrão) lista a pasta; file notification usa fila/eventos da nuvem — melhor para diretórios enormes.
• trigger(availableNow=True) = processa tudo que chegou e encerra: incremental, com custo de batch.

Como decidir na prova

• "Banco relacional operacional, inserts/updates/deletes, mínimo de código" → Lakeflow Connect gerenciado (CDC).
• "Milhões de arquivos chegando, schema muda" → Auto Loader.
• "Carga SQL simples e idempotente de alguns milhares de arquivos" → COPY INTO.
• "Arquivo único do analista" → upload pela UI.

Medalhão em 30 segundos

Bronze = cru, como chegou. Silver = limpo, tipado, deduplicado. Gold = agregado para o negócio.

Bronze → Silver em PySpark

from pyspark.sql import functions as F

silver = (spark.table("main.bronze.pedidos")
  .withColumn("valor", F.col("valor").cast("decimal(12,2)"))
  .withColumn("data_pedido", F.to_date("data_pedido", "yyyy-MM-dd"))
  .fillna({"canal": "desconhecido"})
  .filter(F.col("pedido_id").isNotNull())
  .dropDuplicates(["pedido_id"]))

silver.write.mode("overwrite").saveAsTable("main.silver.pedidos")

Joins e unions que caem

Operação	O que faz
a.join(b, "id")	inner: só chaves dos dois lados
a.join(b, "id", "left")	tudo de A; nulos onde B não casa
a.join(F.broadcast(b), "id")	copia B (pequena) para todos os executores → sem shuffle de A
a.join(b, ["id", "data"])	múltiplas chaves
a.crossJoin(b)	produto cartesiano (cuidado!)
a.union(b) / unionByName	junta linhas por posição / por nome. Em DataFrame, union não remove duplicatas
SQL UNION vs UNION ALL	UNION remove duplicatas; UNION ALL mantém

Colunas e arrays

df.withColumnRenamed("nm", "nome") \
  .withColumn("partes", F.split("endereco", ",")) \
  .withColumn("item", F.explode("itens")) \
  .drop("coluna_lixo")

Agregações

count, avg, approx_count_distinct (HyperLogLog: rápido, erro pequeno), df.summary() / df.describe() (estatísticas). Deduplicação: dropDuplicates([...]) ou ROW_NUMBER() OVER (PARTITION BY chave ORDER BY ts DESC) = 1.

Parâmetros de tuning

• spark.sql.shuffle.partitions (padrão 200): nº de partições depois de shuffles (joins/groupBy). Com AQE, o Spark ajusta sozinho.
• spark.sql.autoBroadcastJoinThreshold (padrão 10 MB): tabelas menores que isso são broadcast automaticamente.
• spark.executor.memory / spark.driver.memory: memória; collect() grande estoura o driver. Sempre meça de novo depois de mudar (Spark UI / query profile).

Objetos gold

Objeto	Armazena dados?	Atualização	Uso
View	Não (só a query)	Sempre atual, recalcula a cada leitura	Lógica leve, segurança
Materialized view	Sim	Refresh (agendado/pipeline), incremental quando possível	Agregações caras lidas por BI
Streaming table	Sim	Processa só o que é novo (append)	Ingestão incremental/baixa latência
Tabela	Sim	Você escreve	Controle total

Qualidade

Em pipelines declarativos, expectations:

CREATE OR REFRESH STREAMING TABLE silver_pedidos (
  CONSTRAINT id_ok EXPECT (pedido_id IS NOT NULL) ON VIOLATION DROP ROW,
  CONSTRAINT valor_ok EXPECT (valor >= 0) ON VIOLATION FAIL UPDATE
) AS SELECT * FROM STREAM(bronze_pedidos);

Sem ON VIOLATION: só registra a métrica (warn). Fora de pipelines: CHECK constraints em tabelas Delta.

Anatomia de um job

Um job é um DAG de tarefas. Cada tarefa tem tipo (notebook, Python script/wheel, SQL query, SQL file, dashboard, pipeline, dbt, Run Job, If/else, For each), compute, parâmetros e dependências (depends_on).

Controle de fluxo

• Retries: por tarefa (max_retries, intervalo). Bom para falhas transitórias.
• If/else condition: compara valores (ex.: {{tasks.validar.values.qtd_linhas}} > 0) e segue por um ramo.
• For each: executa uma tarefa aninhada para cada item de uma lista (ex.: lista de países), com concorrência configurável.
• Run if: quando uma tarefa roda em relação às anteriores — ALL_SUCCESS (padrão), AT_LEAST_ONE_SUCCESS, NONE_FAILED, ALL_DONE, AT_LEAST_ONE_FAILED, ALL_FAILED. Ex.: tarefa de alerta com AT_LEAST_ONE_FAILED.
• Task values: uma tarefa publica (dbutils.jobs.taskValues.set("qtd", 10)) e outra lê.

Gatilhos

Gatilho	Dispara quando	Use se
Scheduled (cron)	Horário fixo	O dado chega sempre no mesmo horário
File arrival	Chegam arquivos novos num local do UC/volume	O dado chega em horários imprevisíveis
Table update	Tabelas de origem recebem commits	O job depende de outra tabela/pipeline
Continuous	Mantém sempre uma execução ativa	Streaming contínuo

Regra: se o horário da chegada é incerto, gatilho por dados evita rodar vazio (desperdício) ou rodar cedo demais (dado incompleto).

Git folders

Clone de um repositório Git dentro do workspace. Pela UI você cria/troca branch, faz commit & push, pull e abre o PR no provedor (GitHub, Azure DevOps, GitLab…). Cada pessoa trabalha no seu Git folder e na sua branch.

Declarative Automation Bundles (DABs)

Infra + código como arquivos versionados. O coração é o databricks.yml:

bundle:
  name: vendas_etl

variables:
  catalogo:
    default: dev_catalog

resources:
  jobs:
    vendas_diario:
      name: vendas-diario-${bundle.target}
      tasks:
        - task_key: ingestao
          notebook_task:
            notebook_path: ./src/ingestao.py
            base_parameters:
              catalogo: ${var.catalogo}

targets:
  dev:
    mode: development       # prefixa recursos com [dev usuario], pausa agendas
    default: true
  prod:
    mode: production
    variables:
      catalogo: prod_catalog
    workspace:
      host: https://empresa-prod.cloud.databricks.com

O fluxo na CLI

databricks bundle validate -t prod   # checa sintaxe e referências
databricks bundle deploy   -t prod   # cria/atualiza jobs, pipelines, etc.
databricks bundle run vendas_diario -t prod
databricks bundle destroy  -t dev    # remove o que o bundle criou

Num pipeline de CI (GitHub Actions, Azure DevOps): PR → validate + testes → merge → deploy -t prod com um service principal.

Por que bundles e não "copiar notebook"

Mesmo código em todos os ambientes, diferenças só nos targets/variáveis, histórico no Git, deploy reproduzível e revisável.

Lendo a Spark UI

• Skew: na aba Stages, Summary Metrics com Max muito maior que Median (ex.: median 400 MB, max 5 GB de shuffle read) → uma task demora muito mais que as outras. Solução: AQE com skew join (ligado por padrão), salting, ou filtrar a chave quente.
• Shuffle alto: muitos dados trocados entre executores (joins/groupBy). Reduza com broadcast join, filtrando antes, ou Liquid Clustering.
• Spill (memory/disk): partição não cabe na memória → mais partições, mais memória, ou corrigir o skew.

Histórico do job

A visão de runs mostra duração por execução: compare a execução atual com a mediana histórica. O grafo mostra qual tarefa falhou/atrasou e quem ficou esperando ("upstream blocker").

Liquid Clustering e Predictive Optimization

CREATE TABLE main.silver.eventos (...) CLUSTER BY (cliente_id, data_evento);
ALTER TABLE main.silver.eventos CLUSTER BY AUTO;   -- a Databricks escolhe as chaves
OPTIMIZE main.silver.eventos;                       -- reorganiza incrementalmente

• Substitui partição + ZORDER; chaves podem mudar sem reescrever a tabela.
• Predictive optimization (tabelas gerenciadas do UC) roda OPTIMIZE, VACUUM e ANALYZE automaticamente quando vale a pena.

Erros clássicos

Sintoma	Causa provável	O que fazer
Cluster não sobe	Cota/capacidade da nuvem, init script falhando, permissão	Ver event log do cluster e logs do init script
ModuleNotFoundError / versão errada	Conflito de bibliotecas	Fixar versões, usar environments do serverless, instalar no escopo certo
OutOfMemoryError no driver	collect()/toPandas() grande, broadcast enorme	Evitar trazer dados ao driver, aumentar memória do driver
OOM / spill no executor	Partições grandes ou skew	Mais partições, AQE, instância com mais memória

Gerenciada x externa

	Gerenciada	Externa
Onde ficam os arquivos	Local gerenciado pelo UC (do catálogo/schema)	Um LOCATION que você define
DROP TABLE	Remove metadados e dados (recuperáveis por 7 dias com UNDROP)	Remove só os metadados; arquivos ficam
Otimizações automáticas	Sim (predictive optimization)	Não

CREATE TABLE main.silver.clientes (id BIGINT, nome STRING);            -- gerenciada
CREATE TABLE main.raw.logs (linha STRING) LOCATION 's3://bucket/logs';  -- externa
ALTER TABLE main.raw.logs SET MANAGED;                                  -- externa → gerenciada

Hierarquia e privilégios

metastore → catálogo → schema → tabela/view/volume/função. Para ler uma tabela é preciso:

GRANT USE CATALOG ON CATALOG main TO `analistas`;
GRANT USE SCHEMA  ON SCHEMA  main.gold TO `analistas`;
GRANT SELECT      ON SCHEMA  main.gold TO `analistas`;   -- herdado por todas as tabelas, inclusive futuras
REVOKE SELECT ON TABLE main.gold.salarios FROM `analistas`;

Privilégios concedidos num nível são herdados pelos objetos abaixo. Conceda a grupos, não a pessoas. Jobs em produção rodam como service principal.

Filtros de linha e máscaras de coluna

CREATE FUNCTION main.seg.so_minha_regiao(regiao STRING)
RETURN is_account_group_member('admins') OR regiao = current_user_region();

ALTER TABLE main.gold.vendas SET ROW FILTER main.seg.so_minha_regiao ON (regiao);

CREATE FUNCTION main.seg.mascara_email(email STRING)
RETURN CASE WHEN is_account_group_member('rh') THEN email ELSE '***' END;

ALTER TABLE main.gold.clientes ALTER COLUMN email SET MASK main.seg.mascara_email;

ABAC

Em vez de aplicar a máscara tabela por tabela, você marca colunas com governed tags (ex.: pii = email) e cria uma política no catálogo/schema: "toda coluna com a tag pii usa a máscara X para quem não é do grupo Y". Vale para tabelas atuais e futuras.