Databricks Certified Associate Developer for Apache Spark

Quem faz o quê

• Driver: roda seu programa, cria o plano, divide em jobs/stages/tasks e agenda.
• Executores (nos workers): executam tasks e guardam dados em cache. Cada core executa uma task por vez.
• Cluster manager: entrega recursos (standalone, YARN, Kubernetes; na Databricks, gerenciado).

Hierarquia de execução

Application → Job (uma por action) → Stage (fronteira = shuffle) → Task (uma por partição).

Lazy evaluation

Transformations (select, filter, withColumn, join, groupBy) só constroem o plano. Actions (count, collect, show, take, write) disparam a execução. Isso deixa o Catalyst otimizar o plano inteiro (ex.: empurrar filtros para a leitura).

• Narrow (sem shuffle): filter, select, map.
• Wide (com shuffle): groupBy, join, distinct, orderBy, repartition.

Partições

spark.sql.shuffle.partitions (padrão 200) define quantas partições saem de um shuffle. repartition(n) faz shuffle e pode aumentar/diminuir; coalesce(n) só diminui, sem shuffle completo.

Cache e storage levels

df.cache() = persist(StorageLevel.MEMORY_AND_DISK) para DataFrames (no PySpark recente aparece como MEMORY_AND_DISK_DESER, a variante desserializada). Outras: MEMORY_ONLY, DISK_ONLY, _SER, _2 (replicado). O cache também é lazy: só materializa na próxima action. unpersist() libera.

GC

Muito objeto na JVM → pausas de garbage collection (veja "GC time" na Spark UI). Ajuda: menos dados em cache, formatos serializados, partições menores, evitar UDFs Python que criam muitos objetos.

Vantagens e desafios

Processamento distribuído em memória, APIs unificadas (SQL, Python, streaming, ML), tolerância a falhas por linhagem. Desafios: tuning (memória, partições, skew), custo de shuffle, curva de aprendizado.

Ler e escrever

df = (spark.read.format("csv").option("header", True).schema("id INT, pais STRING, valor DOUBLE")
        .load("/data/vendas.csv"))
df.write.mode("overwrite").partitionBy("pais").parquet("/data/out")

jdbc = (spark.read.format("jdbc")
          .option("url", "jdbc:postgresql://db:5432/erp").option("dbtable", "public.clientes")
          .option("user", u).option("password", p).load())

Save modes

Modo	Se o destino existe
error / errorifexists (padrão)	Falha
append	Acrescenta
overwrite	Substitui
ignore	Não faz nada

SQL direto em arquivos

SELECT * FROM parquet.`/data/out`;
SELECT * FROM json.`/data/eventos/*.json`;
SELECT * FROM delta.`/data/tabela_delta`;

Tabelas persistentes

df.write.partitionBy("ano").sortBy("id").bucketBy(8, "id").saveAsTable("vendas")   # sortBy exige bucketBy

saveAsTable registra no metastore/catálogo; continua existindo depois da sessão.

Views temporárias

df.createOrReplaceTempView("vendas_v")          # só nesta SparkSession
df.createOrReplaceGlobalTempView("vendas_g")    # global_temp.vendas_g, todas as sessões da aplicação
spark.sql("SELECT pais, SUM(valor) FROM vendas_v GROUP BY pais").show()

Colunas

from pyspark.sql import functions as F

df = (df.withColumn("state", F.col("division")).drop("division")
        .withColumnRenamed("mName", "managerName")
        .withColumn("partes", F.split("endereco", ","))
        .withColumn("tag", F.explode("tags"))
        .filter((F.col("valor") > 0) & F.col("pais").isin("BR", "PT")))

Nulos e duplicatas

df.na.drop() (= how="any": remove linha com qualquer nulo); na.drop("all") só se todas forem nulas; subset=[...]; na.fill({"col": 0}); dropDuplicates(["id"]); distinct().

Agregações

df.groupBy("pais").agg(F.count("*").alias("n"), F.avg("valor"), F.approx_count_distinct("usuario"))
df.summary().show()   # count, mean, stddev, min, 25%, 50%, 75%, max

approx_count_distinct usa HyperLogLog: muito mais barato que countDistinct, com erro pequeno configurável.

Datas

df.withColumn("data", F.from_unixtime("epoch", "yyyy-MM-dd")) \
  .withColumn("ts", F.to_timestamp(F.from_unixtime("epoch"))) \
  .withColumn("ano", F.year("ts")).withColumn("mes", F.month("ts")).withColumn("dia_semana", F.dayofweek("ts"))

Joins e unions

a.join(b, ["id", "data"], "left"), a.crossJoin(b), a.join(F.broadcast(b), "id"). union (= UNION ALL, por posição), unionByName (por nome; allowMissingColumns=True).

Ordenar, schema, listas

linhas = df.orderBy(F.col("salary").desc(), F.col("age").desc()).collect()   # lista de Row
df.printSchema()
lista = [r.asDict() for r in linhas]
for r in df.toLocalIterator(): ...      # itera sem trazer tudo de uma vez

UDFs

@F.udf("string")
def normaliza(s): return s.strip().lower() if s else None
df.withColumn("nome_n", normaliza("nome"))
spark.udf.register("normaliza", normaliza)   # usar em SQL

Prefira funções nativas (mais rápidas). UDF sem estado aplica lógica por linha; lógica com estado entre eventos (sessões, totais acumulados) usa operadores com estado de streaming (applyInPandasWithState, transformWithState) que guardam dados no StateStore.

Broadcast e acumuladores

• Broadcast variable: valor somente leitura enviado uma vez por executor (ex.: dicionário de lookup).
• Accumulator: contador que os executores só somam e o driver lê (ex.: contar registros inválidos).
• Broadcast join: a tabela pequena vai inteira para cada executor → a grande não precisa de shuffle. Automático abaixo de spark.sql.autoBroadcastJoinThreshold (10 MB) ou com F.broadcast().

Estratégias

• Poucas partições → cores ociosos; muitas → overhead. Regra prática: partições de ~100–200 MB, pelo menos 2–3× o número de cores.
• Skew: uma chave com muito mais dados → uma task lenta. Veja Max ≫ Median na Spark UI. Soluções: AQE skew join, salting, broadcast.
• Menos shuffle: filtre e selecione cedo, broadcast de tabelas pequenas, evite distinct/orderBy desnecessários.

AQE (ligado por padrão desde o Spark 3.2)

Reotimiza o plano em tempo de execução com estatísticas reais:

1. Coalesce de partições pós-shuffle (junta partições pequenas);
2. Troca de estratégia de join (sort-merge → broadcast quando um lado ficou pequeno);
3. Skew join (divide partições gigantes).

Logs

Logs de executor e driver pela Spark UI (aba Executors → stdout/stderr) ou pelos logs do cluster. spark.sparkContext.setLogLevel("WARN") ajusta o nível; use log4j2 para customizar. Sinais: OutOfMemoryError no driver (collect grande), executor lost/OOM (partições grandes, skew), CPU ociosa (poucas partições).

Modelo

O stream é uma tabela sem fim; sua query roda incrementalmente em micro-batches. Checkpoint (offsets + estado) + fonte reprocessável + sink idempotente = exactly-once.

eventos = (spark.readStream.format("json").schema(schema).load("/data/in"))

contagem = (eventos.withWatermark("ts", "10 minutes")
              .groupBy(F.window("ts", "15 minutes", "5 minutes"), "produto").count())

(contagem.writeStream.outputMode("update").format("console")
   .option("checkpointLocation", "/chk/contagem").trigger(processingTime="2 minutes").start())

Output modes

append (só linhas novas/finais), update (linhas que mudaram), complete (resultado inteiro — só com agregação).

Sinks

File (parquet, json…), Kafka, foreach / foreachBatch, console e memory (debug), tabelas (toTable).

Deduplicação

eventos.dropDuplicates(["evento_id"])                                   # estado cresce para sempre
eventos.withWatermark("ts", "1 hour").dropDuplicates(["evento_id", "ts"])  # estado limitado pela watermark
eventos.withWatermark("ts", "1 hour").dropDuplicatesWithinWatermark(["evento_id"])

Spark Connect

Arquitetura cliente-servidor: o cliente leve (Python, Scala, Go…) envia planos lógicos via gRPC para o servidor Spark remoto. Vantagens: rodar Spark a partir de IDEs, notebooks e apps sem JVM local; isolamento (um cliente que trava não derruba o driver); upgrades independentes. SparkSession.builder.remote("sc://host:15002").getOrCreate(). Na Databricks, é a base do Databricks Connect e do serverless.

Modos de deploy

Modo	Onde fica o driver
Client	Na máquina que submeteu (bom para interativo)
Cluster	Dentro do cluster (bom para produção; independe da máquina do usuário)
Local	Tudo em uma JVM numa máquina; executores são threads (local[*]) — testes e desenvolvimento

Pandas API on Spark

import pyspark.pandas as ps
psdf = ps.read_csv("/data/vendas.csv")
psdf.groupby("pais")["valor"].mean().sort_values()
psdf.to_spark()     # volta a ser DataFrame do Spark; df.pandas_api() faz o caminho inverso

Sintaxe do pandas com execução distribuída — migração de código pandas para dados que não cabem numa máquina.

Pandas UDF

import pandas as pd
from pyspark.sql.functions import pandas_udf

@pandas_udf("double")
def com_imposto(valor: pd.Series) -> pd.Series:
    return valor * 1.1

df.withColumn("valor_final", com_imposto("valor"))

Vetorizada com Apache Arrow: processa lotes inteiros, muito mais rápida que UDF Python linha a linha.