Estudo Aprofundado

Recursos gerais

01Pratique no Databricks Free Edition ou localmente com pip install pyspark

02Sempre que uma query estiver lenta, rode df.explain(True) antes de otimizar às cegas

03Prefira a DataFrame API a RDDs — é otimizada pelo Catalyst e mais legível

04Consulte a API reference oficial para a assinatura exata de cada função antes de usar

PySpark API Reference PySpark on Databricks Free Edition Funções SQL do Spark

Fundamentos

Como o Spark pensa antes de escrever a primeira linha

SparkSession e arquiteturaBásico

Driver, executors e cluster manager

Todo programa PySpark começa por uma SparkSession, o ponto de entrada único para DataFrame, SQL e streaming. O driver planeja o trabalho e distribui tarefas para os executors, que rodam em paralelo nos nós do cluster. No Databricks a sessão (spark) já vem criada automaticamente em cada notebook.

TeoriaUm cluster Spark segue o modelo master-worker: o driver roda seu código, mantém o grafo de execução (DAG) e negocia recursos com um cluster manager (Standalone, YARN, Kubernetes ou, no Databricks, o próprio gerenciador do workspace). Os executors são processos JVM de longa duração nos nós de trabalho, cada um com um número fixo de núcleos e memória, responsáveis por rodar as tarefas e guardar dados em cache. Toda ação dispara um Job, que o driver quebra em Stages nos pontos de shuffle, e cada stage em Tasks — uma por partição — distribuídas entre os executors disponíveis.

💡Onde usar: é o primeiro código de qualquer script ou notebook PySpark, fora do Databricks (execução local ou em cluster próprio).

from pyspark.sql import SparkSession

spark = (SparkSession.builder
  .appName("estudo-pyspark")
  .getOrCreate())

# No Databricks, `spark` já existe pronto no notebook — não precisa criar

Cluster Mode Overview PySpark on Databricks SparkSession — API Reference

RDD vs DataFrameBásico

Quando (não) voltar para o RDD

RDD (Resilient Distributed Dataset) é a estrutura de dados original do Spark: uma coleção distribuída sem schema, manipulada com funções Python puras. DataFrame é construído sobre o RDD, tem schema (colunas e tipos) e passa pelo otimizador Catalyst — na prática, roda mais rápido que o RDD equivalente na quase totalidade dos casos.

TeoriaUm RDD é imutável e guarda sua lineage — o histórico de transformações que o originou — em vez de replicar dados; se um executor cai, o Spark simplesmente recalcula a partição perdida a partir da lineage, sem precisar de backup. O DataFrame é construído sobre o RDD, mas adiciona um schema explícito e um plano lógico que o Catalyst Optimizer consegue analisar e reescrever. Em Python isso importa especialmente: como o PySpark não tem o mesmo type-safety em tempo de compilação que Scala/Java (Dataset), é o plano lógico do DataFrame — não o tipo estático — que garante a otimização.

⚠️Onde usar RDD: só quando você precisa de controle de baixo nível sobre particionamento/paralelismo ou dados sem estrutura tabular. Para ETL, analytics e ML, use DataFrame.

# RDD — baixo nível, sem otimizador
rdd = spark.sparkContext.textFile("/dados/log.txt")
contagem = rdd.flatMap(lambda l: l.split(" ")).count()

# DataFrame — schema + Catalyst Optimizer
df = spark.read.json("/dados/log.json")
df.groupBy("nivel").count().show()

RDD Programming Guide Tutorial: Load and transform data com DataFrames

Lazy evaluation e Catalyst OptimizerIntermediário

Transformações vs ações

Transformações (filter, select, join) não executam nada — só constroem um plano lógico. Só quando uma ação (show, count, write) é chamada, o Catalyst Optimizer reescreve o plano lógico em plano físico e o Spark executa de fato.

TeoriaO Catalyst Optimizer processa toda query em quatro fases: análise (resolve nomes de coluna e tabela contra o catálogo), otimização lógica (aplica regras como predicate pushdown, column pruning e constant folding, independente de como os dados estão armazenados), planejamento físico (gera candidatos concretos de execução e escolhe um pelo custo estimado) e geração de código, onde o Tungsten compila o plano final em bytecode JVM (whole-stage code generation), evitando a sobrecarga de milhares de chamadas de função virtual por linha. É essa cadeia — não a linguagem usada para escrever a query — que faz DataFrame e Spark SQL terem a mesma performance.

💡Onde usar: entender isso evita o erro clássico de achar que uma transformação "travou" — ela não executa até uma ação ser chamada.

Transformações: select, filter, withColumn, join, groupBy, orderBy

Ações: show, count, collect, take, write, toPandas

df.explain(True) mostra plano lógico, otimizado e físico lado a lado

Spark SQL, DataFrames and Datasets Guide

DataFrame API

O dia a dia de quem escreve PySpark

Leitura e escrita de dadosBásico

read/write — parquet, delta, json, csv, jdbc

spark.read e df.write são os pontos de entrada e saída de qualquer pipeline. O formato muda o método (.parquet(), .json(), .table()) ou o .format() genérico.

TeoriaFormatos colunares como Parquet e Delta guardam metadados de schema e estatísticas por arquivo, o que permite ao Spark aplicar predicate pushdown (filtrar antes de ler o arquivo inteiro) e column pruning (ler só as colunas necessárias) direto na camada de armazenamento. Formatos baseados em linha como CSV e JSON não têm essa vantagem — cada leitura processa o arquivo inteiro, e usar inferSchema ainda soma uma passada extra só para detectar os tipos. Por isso a recomendação padrão é declarar o schema explicitamente com StructType em produção.

💡Onde usar: no Databricks, prefira sempre delta (formato padrão das tabelas) em vez de CSV/JSON para dados que serão lidos/atualizados repetidamente.

df = (spark.read
  .format("parquet")
  .option("mergeSchema", True)
  .load("/Volumes/raw/vendas/"))

(df.write
  .format("delta")
  .mode("overwrite")
  .saveAsTable("bronze.vendas"))

PySpark basics Delta Lake

Seleção, filtro e colunasBásico

select, filter, withColumn, funções

A maior parte da manipulação de colunas passa pelo módulo pyspark.sql.functions, importado por convenção como F para não sobrescrever funções nativas do Python.

TeoriaUma expressão como F.col("valor") > 100 não é avaliada na hora — ela constrói uma árvore de expressão (um objeto Column) que fica anexada ao plano lógico. É justamente por existir essa árvore, e não um valor Python já calculado, que o Catalyst consegue mover um filtro para antes de um join ou empurrá-lo para dentro do leitor de Parquet — otimizações que seriam impossíveis se a filtragem já tivesse acontecido em Python puro.

import pyspark.sql.functions as F

resultado = (df
  .select("cliente_id", "valor", "data")
  .filter(F.col("valor") > 100)
  .withColumn("ano", F.year("data"))
  .withColumnRenamed("valor", "valor_total"))

select / filter (where) / withColumn / drop / alias / cast

Sempre prefira F.col("nome") em vez de string solta em expressões complexas

pyspark.sql.functions — API Reference

Agregações e window functionsIntermediário

groupBy, agg, Window

groupBy().agg() resume linhas em grupos. Já as window functions calculam algo por linha sem colapsar o DataFrame — essencial para ranking, acumulados e comparação com a linha anterior.

TeoriaEm um groupBy().agg(), o Spark faz uma agregação parcial dentro de cada partição antes do shuffle (equivalente a um combiner do MapReduce), reduzindo o volume de dados que trafega pela rede. Já uma window function não colapsa linhas: ela particiona e ordena os dados (também via shuffle, quando partitionBy é usado) mas mantém uma linha de saída para cada linha de entrada, com o valor da função calculado sobre a janela ao redor dela.

💡Onde usar window functions: "top N por categoria", médias móveis, deduplicação mantendo o registro mais recente por chave.

from pyspark.sql.window import Window
import pyspark.sql.functions as F

# Agregação simples
vendas_por_regiao = df.groupBy("regiao").agg(F.sum("valor").alias("total"))

# Ranking por partição (window function)
janela = Window.partitionBy("regiao").orderBy(F.col("valor").desc())
top3 = (df
  .withColumn("rank", F.row_number().over(janela))
  .filter(F.col("rank") <= 3))

pyspark.sql.Window — API Reference

JoinsIntermediário

Tipos de join e broadcast

Join é a operação mais cara de um pipeline Spark porque normalmente exige shuffle (redistribuir dados entre executors). Quando um dos lados é pequeno, um broadcast join evita o shuffle inteiramente.

TeoriaO planejador físico escolhe entre três estratégias principais: Broadcast Hash Join (copia o lado pequeno inteiro para todos os executors, elimina o shuffle do lado grande), Shuffle Hash Join e Sort-Merge Join — a estratégia padrão quando os dois lados são grandes, exige shuffle e ordenação de ambos pela chave de join. A escolha considera estatísticas da tabela (tamanho, cardinalidade) quando disponíveis via ANALYZE TABLE, e o AQE pode revisar essa escolha em tempo de execução com números reais.

⚠️Cuidado: broadcast só compensa para tabelas pequenas (tipicamente até algumas centenas de MB). Forçar broadcast numa tabela grande pode estourar a memória do driver.

from pyspark.sql.functions import broadcast

# Join padrão (inner, left, right, outer, semi, anti)
resultado = pedidos.join(clientes, "cliente_id", "left")

# Broadcast join — clientes é pequena, evita shuffle
resultado = pedidos.join(broadcast(clientes), "cliente_id")

Adaptive Query Execution (converte joins automaticamente)

Tipos complexos e dados aninhadosIntermediário

structs, arrays, maps, explode

Dados vindos de APIs e sistemas de origem raramente são planos: struct representa um objeto aninhado com campos nomeados, array uma lista de valores do mesmo tipo, e map um dicionário chave-valor — os três aparecem o tempo todo ao ler JSON. O acesso a um campo de struct usa notação de ponto (coluna.campo), e funções como explode, array_contains e as higher-order functions (transform, filter) operam sobre arrays e maps sem precisar "achatar" o DataFrame manualmente em Python.

TeoriaTipos complexos fazem parte do StructType do schema como qualquer outro tipo — não são um recurso à parte, e o Catalyst consegue fazer pruning até de campos individuais dentro de um struct ao ler Parquet/Delta (nested column pruning), lendo só os campos realmente usados. explode é uma transformação especial: para cada linha com um array de N elementos, ela gera N linhas de saída — é a única operação comum da DataFrame API que aumenta o número de linhas.

⚠️Cuidado: explode numa coluna com arrays de tamanho muito variável (um pedido com 2 itens, outro com 50 mil) é uma fonte clássica de skew — a linha do array gigante vira uma partição desproporcional depois do explode.

import pyspark.sql.functions as F

df2 = (df
  .select("pedido_id", "cliente.nome", "cliente.email")   # dot notation em struct
  .withColumn("item", F.explode("itens"))                    # 1 linha por item do array
  .withColumn("tem_desconto", F.array_contains("tags", "promo")))

pyspark.sql.functions — API Reference Spark SQL, DataFrames and Datasets Guide

Spark SQL

Misturando SQL com PySpark no mesmo pipeline

Spark SQL e temp viewsBásico

spark.sql(), createOrReplaceTempView

Qualquer DataFrame pode virar uma view temporária e ser consultado com SQL puro — os dois caminhos (DataFrame API e SQL) passam pelo mesmo Catalyst Optimizer, então não há perda de performance ao escolher um ou outro.

TeoriaUma temp view não materializa dado nenhum — é só um nome apontando para o mesmo plano lógico do DataFrame que a originou, válido apenas na sessão atual (ou entre sessões, se for uma global temp view, registrada no banco especial global_temp). O texto SQL passa por um parser próprio do Spark (baseado em ANTLR) que gera exatamente o mesmo tipo de plano lógico não resolvido que as chamadas da DataFrame API geram — por isso as duas formas de escrever a mesma query têm performance idêntica.

💡Onde usar: quando o time tem mais familiaridade com SQL, ou para reaproveitar queries já escritas por analistas dentro de um pipeline PySpark.

df.createOrReplaceTempView("vendas_tmp")

resultado = spark.sql("""
    SELECT regiao, SUM(valor) AS total
    FROM vendas_tmp
    GROUP BY regiao
    ORDER BY total DESC
""")

Spark SQL, DataFrames and Datasets Guide

Performance e Otimização

Onde a maioria dos problemas de PySpark realmente mora

ParticionamentoIntermediário

repartition, coalesce, partition pruning

repartition() faz shuffle completo para redistribuir dados (aumenta ou diminui partições). coalesce() só junta partições existentes, sem shuffle — muito mais barato para reduzir o número de arquivos de saída.

TeoriaO número de partições define o grau de paralelismo: poucas partições deixam núcleos do cluster ociosos, partições demais somam overhead de agendamento de tarefas. Todo shuffle (join, groupBy, repartition) usa o valor de spark.sql.shuffle.partitions (200 por padrão) para decidir em quantas partições reparticionar os dados — um número fixo, pensado para um volume médio de dados, que raramente é ideal para todo tipo de carga. O recurso coalesce automático de partições pós-shuffle do AQE existe justamente para corrigir esse valor fixo em tempo real, sem que você precise acertar o número manualmente.

💡Onde usar coalesce: no fim de um pipeline, antes de escrever, para evitar gerar centenas de arquivos pequenos.

# Redistribui em 200 partições por cliente_id (shuffle)
df2 = df.repartition(200, "cliente_id")

# Reduz para 8 arquivos de saída sem shuffle
df2.coalesce(8).write.format("delta").save("/tmp/saida")

Spark Tuning Guide Optimization recommendations on Databricks

Caching e persistênciaIntermediário

cache, persist, storage levels

Como o Spark é lazy, um DataFrame usado várias vezes seria recalculado do zero a cada ação. cache() guarda o resultado em memória (com fallback em disco) para reaproveitar entre ações.

Teoriacache() é um atalho para persist(StorageLevel.MEMORY_AND_DISK): tenta guardar em memória e usa disco como fallback quando não cabe. Outros níveis (MEMORY_ONLY, DISK_ONLY, variantes _SER serializadas) trocam velocidade de acesso por uso de memória. Como o Spark também é lazy para cache, nada é de fato armazenado até uma ação materializar o DataFrame — por isso é comum ver um count() logo após o cache(), só para forçar essa materialização.

⚠️Cuidado: cache só compensa quando o DataFrame é reutilizado várias vezes. Sempre libere com .unpersist() quando não precisar mais — memória de cache não liberada é uma causa comum de OOM em cluster.

df_base = spark.read.table("silver.pedidos").cache()

df_base.count()               # materializa o cache
total_a = df_base.filter("regiao = 'A'").count()
total_b = df_base.filter("regiao = 'B'").count()

df_base.unpersist()             # libera a memória

RDD Persistence (storage levels) Optimization recommendations on Databricks

Explain plans e AQEAvançado

Adaptive Query Execution

df.explain(True) mostra o plano lógico, o plano lógico otimizado e o plano físico de uma query. O Adaptive Query Execution (AQE), ativo por padrão, reotimiza o plano em tempo de execução usando estatísticas reais coletadas após cada shuffle — troca sort-merge join por broadcast join, ajusta o número de partições pós-shuffle e trata skew automaticamente.

TeoriaO Spark divide a query em query stages, delimitados por pontos de shuffle ou broadcast (chamados de exchange). Ao final de cada stage, o Spark já tem estatísticas reais — não estimadas — do volume de dados produzido, e o AQE usa esses números para replanejar o que falta executar: pode trocar um sort-merge join por broadcast join se um lado acabou menor que o esperado, redefinir o número de partições pós-shuffle, ou dividir uma partição de dados desbalanceada (skew) em pedaços menores.

💡Onde usar: antes de tunar manualmente qualquer query lenta, leia o plano físico — muitas vezes o problema é um filtro que não desceu (predicate pushdown) ou um shuffle desnecessário.

df.explain(True)   # plano lógico, otimizado e físico

# AQE já vem ativo por padrão; para conferir:
spark.conf.get("spark.sql.adaptive.enabled")

Adaptive query execution Spark Tuning Guide

Shuffle, skew e broadcastAvançado

Por que o job trava em uma tarefa só

Shuffle é a redistribuição de dados entre executors (em join, groupBy, repartition) — a operação mais cara do Spark, com I/O de disco e rede. Skew acontece quando uma chave concentra dados demais, deixando uma única tarefa muito mais lenta que as outras. É a causa nº 1 de "job travado em 199/200 tasks".

TeoriaNo shuffle, o lado do map escreve arquivos locais em disco, particionados pelo hash da chave; o lado do reduce busca essas partições pela rede via um serviço de shuffle. Skew acontece quando uma chave concentra desproporcionalmente mais linhas que as outras — a tarefa responsável por essa partição fica muito maior que as demais e vira o gargalo de todo o stage. A otimização de skew do AQE resolve isso automaticamente na maioria dos casos, dividindo a partição grande em sub-partições e replicando o lado correspondente do join; salting manual da chave (adicionar um sufixo aleatório) é a saída quando o AQE não é suficiente.

⚠️Sintoma clássico de skew: no Spark UI, uma task de um stage demora minutos enquanto todas as outras terminam em segundos. AQE já mitiga boa parte automaticamente; salting manual da chave é o último recurso.

# Salting manual para quebrar uma chave muito concentrada
import pyspark.sql.functions as F

df_salgado = df.withColumn(
    "chave_salgada",
    F.concat(F.col("chave"), F.lit("_"), (F.rand() * 10).cast("int"))
)

Spark Tuning Guide Adaptive query execution (trata skew automaticamente)

Variáveis compartilhadasIntermediário

Broadcast variables e Accumulators

Além do broadcast join automático, você pode fazer o broadcast manual de qualquer valor somente-leitura — um dicionário Python pequeno, por exemplo — para que ele seja copiado uma única vez para cada executor. Accumulators são o caminho inverso: contadores de escrita-only que cada executor incrementa localmente e que o driver soma automaticamente no final, úteis para métricas simples como "quantas linhas vieram com um campo inválido".

TeoriaUm broadcast (sc.broadcast(valor)) é serializado uma única vez e fica em cache em cada executor pelo tempo de vida da aplicação — diferente de uma variável Python comum referenciada dentro de uma UDF, que seria serializada de novo a cada tarefa. Accumulators só têm garantia de valor correto quando incrementados dentro de ações; dentro de transformações, o Spark pode reexecutar uma tarefa (por falha ou execução especulativa) e incrementar o accumulator mais de uma vez para a mesma linha.

🚨Nunca use um accumulator para decidir um resultado de negócio (ex.: "processei X linhas, logo o job está correto") — use-o só como métrica auxiliar de observabilidade.

tabela_cambio = sc.broadcast({"USD": 5.10, "EUR": 5.55})
linhas_invalidas = sc.accumulator(0)

@F.udf("double")
def converter(valor, moeda):
    if moeda not in tabela_cambio.value:
        linhas_invalidas.add(1)
        return None
    return valor * tabela_cambio.value[moeda]

RDD Programming Guide — Shared Variables

Gerenciamento de memóriaAvançado

Execution memory, storage memory e spilling

A memória de heap de um executor é dividida pelo Unified Memory Manager entre memória de execução (usada por shuffles, joins, ordenações e agregações) e memória de armazenamento (cache/persist), com fronteira dinâmica entre as duas. Quando uma operação precisa de mais memória de execução do que está disponível, o Spark não quebra na hora — ele derrama (spill) parte dos dados para disco e continua, só que bem mais devagar.

TeoriaPor padrão, cerca de 60% do heap do executor (após reservar uma fração fixa para o próprio Spark) é dividido entre execução e armazenamento via spark.memory.fraction; a fração mínima garantida para cache é controlada por spark.memory.storageFraction. Ativar memória off-heap (spark.memory.offHeap.enabled) tira parte da memória de execução da gestão do garbage collector da JVM, reduzindo pausas de GC em cargas pesadas de shuffle. O Photon vai além: gerencia sua própria memória fora do heap da JVM inteiramente, para os operadores que suporta.

⚠️Um OutOfMemoryError no driver quase sempre vem de collect(), toPandas() ou um broadcast maior do que o esperado trazendo dados demais para um único processo — raramente é falta de memória nos executors.

# Spill visível no Spark UI (aba Stages), nas colunas
# "Shuffle Spill (Memory)" e "Shuffle Spill (Disk)"

spark.conf.get("spark.memory.fraction")          # padrão: 0.6
spark.conf.get("spark.memory.offHeap.enabled")   # padrão: false

Spark Tuning Guide

Lendo o Spark UIIntermediário

DAG, stages, tasks e onde olhar primeiro

A aba Jobs mostra um job por ação disparada; Stages quebra cada job nos pontos de shuffle, com duração, dados lidos/escritos e spill por tarefa; a aba SQL / DataFrame mostra o plano físico com métricas reais sobrepostas em cada operador; e Executors mostra uso de memória e atividade de garbage collection por executor.

TeoriaA aba SQL/DataFrame é normalmente o ponto de partida mais rápido para diagnosticar uma query lenta: é o mesmo plano de explain(True), só que com números reais de execução em vez de estimativas — dá pra ver exatamente em qual operador (um join, uma agregação, um scan) o tempo está concentrado, sem precisar ler o plano físico em texto.

💡Onde usar: antes de tunar qualquer coisa manualmente, abra a aba Stages e ache o stage mais lento; se a variação de duração entre tarefas do mesmo stage for grande, é sinal de skew, não de falta de recursos.

Jobs — visão geral, um job por ação

Stages — shuffle, spill, variação de duração entre tarefas (skew)

SQL / DataFrame — plano físico com métricas reais sobrepostas

Executors — memória, GC e tarefas ativas por executor

Storage — o que está de fato em cache agora

Web UI — Apache Spark

Z-Ordering, Liquid Clustering e compactaçãoAvançado

OPTIMIZE, ZORDER e Liquid Clustering

Escritas frequentes (streaming, jobs incrementais) tendem a gerar muitos arquivos pequenos numa tabela Delta, deixando leituras mais lentas. OPTIMIZE compacta esses arquivos em arquivos maiores; ZORDER BY reorganiza os dados dentro dos arquivos para que valores próximos de uma coluna fiquem nos mesmos arquivos, melhorando o data skipping em filtros por essa coluna.

TeoriaLiquid Clustering é a evolução recomendada pela Databricks para tabelas novas, substituindo tanto o particionamento Hive-style quanto o ZORDER: em vez de exigir uma coluna de partição fixa (que não pode mudar depois) ou uma reescrita completa a cada ZORDER, o Liquid Clustering reagrupa os dados de forma incremental a cada escrita, permitindo trocar as colunas de clustering ao longo do tempo sem reescrever a tabela inteira.

💡A própria Databricks recomenda usar Liquid Clustering em vez de particionamento Hive-style para tabelas Delta novas.

-- Compactação + Z-Order (tabelas existentes)
OPTIMIZE silver.pedidos ZORDER BY (cliente_id);

-- Liquid Clustering (recomendado para tabelas novas)
CREATE TABLE silver.pedidos_v2 CLUSTER BY (cliente_id);

Optimization recommendations on Databricks

PhotonIntermediário

O motor de execução vetorizado do Databricks

Photon é o motor de execução vetorizado do Databricks, escrito em C++ e compatível com as mesmas APIs Spark/DataFrame — não exige reescrever código, só ativar no cluster ou warehouse. Em vez de processar linha a linha na JVM, ele processa lotes de dados em formato colunar usando instruções vetorizadas (SIMD) da CPU.

TeoriaPhoton é usado por padrão nos SQL Warehouses do Databricks e pode ser ligado em clusters de job/all-purpose. Ele acelera os operadores nativos do Spark (scans, filtros, joins, agregações, escrita em Delta) e gerencia memória fora do heap da JVM, evitando pausas de garbage collection — mas não acelera Python UDFs comuns, que continuam rodando no processo Python de sempre, fora do alcance do Photon.

💡Onde usar: praticamente sempre que disponível para SQL/DataFrame API — o ganho vem de graça, sem mudar código. Para lógica em UDF, o ganho continua vindo de trocar Python UDF por Pandas UDF ou função nativa, não do Photon.

Best practices for performance efficiency

Structured Streaming

Dados contínuos tratados como uma tabela infinita

Fundamentos de streamingIntermediário

readStream, writeStream, triggers, checkpoints

A mesma DataFrame API usada em batch funciona em streaming — o Spark trata o stream como uma tabela que cresce infinitamente. O checkpoint guarda o progresso (offsets processados) para garantir tolerância a falhas e semântica exactly-once.

TeoriaO Structured Streaming usa um modelo de micro-batch: em vez de processar evento a evento, o motor trata o stream como uma tabela que recebe novas linhas a cada gatilho (trigger) e reaplica a mesma lógica incremental de uma query batch sobre os dados novos. Um offset log registra até onde cada fonte já foi lida, e um commit log registra quais micro-batches foram processados com sucesso — juntos, garantem semântica exactly-once mesmo se o job cair e reiniciar no meio de um batch, desde que a fonte e o destino sejam compatíveis com replay/idempotência.

💡Onde usar: ingestão contínua de eventos (Kafka, Event Hubs, Auto Loader), CDC e qualquer pipeline que precise refletir dados novos em minutos, não em execuções batch diárias.

stream = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .load("/Volumes/raw/eventos/"))

(stream.writeStream
  .format("delta")
  .option("checkpointLocation", "/Volumes/checkpoints/eventos")
  .trigger(processingTime="30 seconds")
  .toTable("bronze.eventos"))

Structured Streaming Programming Guide Auto Loader

Watermarking e janelasAvançado

Agregações com dados atrasados

Em streaming, eventos podem chegar atrasados. O watermark define até quando o Spark espera dados atrasados antes de descartar o estado de uma janela — sem ele, o estado acumulado em memória cresceria para sempre.

TeoriaPara agregações com janela, o Spark mantém um state store — um armazenamento de chave-valor (RocksDB, recomendado para estado grande) com o estado parcial de cada janela em aberto. O watermark diz ao motor até quanto tempo de atraso ele deve esperar por dados de uma janela antes de fechá-la e descartar o estado; sem essa marca d'água, o state store cresceria indefinidamente, guardando estado de janelas que talvez nunca mais recebam dados.

⚠️Cuidado: um watermark curto demais descarta dados legítimos que chegaram atrasados; um watermark longo demais mantém estado em memória por mais tempo, custando mais recursos.

import pyspark.sql.functions as F

agregado = (stream
  .withWatermark("timestamp_evento", "10 minutes")
  .groupBy(F.window("timestamp_evento", "5 minutes"), "tipo")
  .count())

Configure RocksDB state store Handling Late Data and Watermarking

foreachBatch e múltiplos sinksAvançado

Escrevendo streaming em destinos sem sink nativo

Structured Streaming só tem sink nativo para um conjunto limitado de formatos (Delta, Kafka, console, memory). foreachBatch dá acesso a cada micro-batch como um DataFrame batch comum dentro de uma função Python, liberando o uso de qualquer API batch — grava em múltiplas tabelas, faz upsert via MERGE, escreve em JDBC, ou manda o mesmo batch para destinos diferentes.

TeoriaA função passada para foreachBatch pode ser chamada mais de uma vez para o mesmo batch_id em cenários de falha e reprocessamento — o Spark garante que cada micro-batch é tentado ao menos uma vez, não que a função roda exatamente uma vez. Para semantica exactly-once de ponta a ponta, as escritas dentro da função precisam ser idempotentes: um MERGE que checa o batch_id já processado, em vez de um INSERT simples.

⚠️Sem idempotência dentro do foreachBatch, um reprocessamento após falha pode gravar o mesmo batch duas vezes no destino.

def escrever_batch(df_batch, batch_id):
    (DeltaTable.forName(spark, "silver.pedidos")
        .alias("t")
        .merge(df_batch.alias("s"), "t.id = s.id AND t.batch_id = s.batch_id")
        .whenNotMatchedInsertAll()
        .execute())

stream.writeStream.foreachBatch(escrever_batch).start()

Structured Streaming Programming Guide

PySpark no ecossistema Databricks

UDFs, Pandas API on Spark e Delta Lake

UDFs: Python vs Pandas UDFIntermediário

Quando o Spark nativo não é suficiente

Uma UDF Python roda linha a linha, com serialização entre JVM e Python a cada chamada — lenta. Uma Pandas UDF processa em lotes (batches) usando Apache Arrow, o que a torna até 100x mais rápida.

TeoriaUma UDF Python comum roda linha a linha: cada linha é serializada da JVM, enviada por pipe para um processo Python separado, processada e o resultado volta serializado — um custo que se paga a cada linha e quebra a geração de código (whole-stage codegen) do Tungsten. A Pandas UDF usa Apache Arrow para transferir lotes inteiros de linhas de uma vez, como colunas contíguas de memória, para o processo Python rodar a lógica vetorizada do pandas/NumPy sobre o lote inteiro — daí o ganho de até 100x.

⚠️Databricks recomenda UDF só para lógica que realmente não existe em pyspark.sql.functions — para tudo que já é nativo (soma, string, data), use a função do Spark, não uma UDF.

import pandas as pd
import pyspark.sql.functions as F
from pyspark.sql.types import DoubleType

# Pandas UDF — processa em lote via Arrow
@F.pandas_udf(DoubleType())
def com_desconto(valor: pd.Series) -> pd.Series:
    return valor * 0.9

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

What are user-defined functions (UDFs)? Pandas user-defined functions

Pandas API on SparkBásico

pyspark.pandas — sintaxe pandas, execução distribuída

Permite escrever com a sintaxe do pandas (antigo projeto Koalas) rodando distribuído no Spark por baixo — bom caminho de migração para quem já tem código pandas e precisa escalar sem reescrever tudo em DataFrame API.

TeoriaInternamente, cada objeto pyspark.pandas mantém um InternalFrame que embrulha um DataFrame Spark comum, mas acrescenta uma coluna de ordenação implícita — necessária porque o Spark, ao contrário do pandas, não garante ordem estável entre partições distribuídas. É essa camada que permite emular o comportamento posicional do pandas (como iloc) em cima de um motor que naturalmente não tem noção de posição.

💡Onde usar: notebooks de análise exploratória vindos de pandas puro, quando o dataset cresceu além da memória de um único nó. Para pipelines de produção, a DataFrame API continua sendo mais previsível.

import pyspark.pandas as ps

psdf = ps.read_parquet("/Volumes/raw/vendas/")
psdf["margem"] = psdf["receita"] - psdf["custo"]
print(psdf.groupby("regiao")["margem"].mean())

Pandas API on Spark — User Guide

Delta Lake com PySparkIntermediário

MERGE, time travel, OPTIMIZE

O formato padrão de tabela no Databricks. A biblioteca delta-spark expõe DeltaTable, usado para upsert (MERGE), viajar no tempo entre versões e compactar arquivos pequenos.

TeoriaO transaction log (pasta _delta_log, uma sequência de arquivos JSON e checkpoints Parquet) registra cada commit feito na tabela e é a fonte única de verdade sobre seu estado atual. As garantias ACID vêm de controle de concorrência otimista: um writer lê a versão mais recente, escreve seus arquivos de dados, e só confirma o commit depois de checar se nenhum outro writer commitou uma mudança conflitante nesse intervalo — se houve conflito, a escrita falha e pode ser repetida, em vez de corromper a tabela.

💡Onde usar MERGE: upsert incremental (evita reescrever a tabela inteira todo dia) e Change Data Capture / SCD Type 2.

from delta.tables import DeltaTable

tabela = DeltaTable.forName(spark, "silver.clientes")

(tabela.alias("t")
  .merge(novos_dados.alias("s"), "t.id = s.id")
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .execute())

# Time travel
spark.read.format("delta").option("versionAsOf", 3).table("silver.clientes")

What are ACID guarantees on Databricks? Delta Lake Optimization recommendations (OPTIMIZE, Z-Order)

Schema evolution no Delta LakeIntermediário

mergeSchema, overwriteSchema e schema enforcement

Por padrão, o Delta aplica schema enforcement: uma escrita com uma coluna nova ou tipo incompatível falha, protegendo a tabela contra corrupção silenciosa. mergeSchema (adiciona colunas novas automaticamente num append) e overwriteSchema (substitui o schema inteiro num overwrite) relaxam essa proteção quando a evolução é esperada.

TeoriaEm streaming, o Auto Loader tem seu próprio controle de evolução de schema: o modo addNewColumns (padrão) reinicia o stream automaticamente ao encontrar uma coluna nova, aplicando o novo schema a partir dali; rescue guarda qualquer dado que não bate com o schema esperado numa coluna à parte (_rescued_data) em vez de falhar ou descartar; failOnNewColumns e none dão mais controle explícito para pipelines onde uma mudança de schema deveria virar um evento, não ser absorvida silenciosamente.

⚠️mergeSchema=true aceita colunas novas sem perguntar — ótimo para flexibilidade, mas pode mascarar uma mudança de schema upstream que deveria virar um alerta. Em pipelines críticos, prefira schema explícito + o modo rescue do Auto Loader a um mergeSchema cego.

(df.write
  .format("delta")
  .option("mergeSchema", True)
  .mode("append")
  .saveAsTable("silver.pedidos"))

# Auto Loader — evolução de schema em streaming
(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .load("/Volumes/raw/pedidos/"))

Delta Lake Auto Loader

Testando pipelines PySparkAvançado

pytest, fixtures de SparkSession e dados de exemplo

Transformações PySpark são mais fáceis de testar quando são funções puras — recebem um DataFrame e devolvem outro, sem ler ou escrever tabelas por dentro. Isso permite testar a lógica com pytest e dados de exemplo pequenos, sem precisar de um cluster real nem de dados de produção.

TeoriaUma fixture de sessão (scope="session") cria uma SparkSession local (master("local[*]")) uma única vez para toda a suíte de testes, evitando o custo de inicializar o Spark a cada teste. Para comparar DataFrames, o Spark 4.x já traz nativamente pyspark.testing.assertDataFrameEqual; bibliotecas como chispa fazem o mesmo com mensagens de diff mais legíveis quando o teste falha, mostrando exatamente quais linhas e colunas divergem.

💡Onde usar: sempre que uma função de transformação tiver alguma regra de negócio não trivial (deduplicação, cálculo, regra condicional) — é o tipo de lógica que quebra silenciosamente numa refatoração futura sem teste.

# tests/conftest.py
import pytest
from pyspark.sql import SparkSession

@pytest.fixture(scope="session")
def spark():
    return SparkSession.builder.master("local[*]").getOrCreate()

# tests/test_transforms.py
from pyspark.testing import assertDataFrameEqual
from meu_pacote.transforms import com_desconto

def test_com_desconto(spark):
    entrada = spark.createDataFrame([(100.0,)], ["valor"])
    esperado = spark.createDataFrame([(90.0,)], ["valor_final"])
    assertDataFrameEqual(com_desconto(entrada), esperado)

Unit testing for Databricks notebooks Run Python tests — Databricks extension for VS Code