Performance

A maioria dos pipelines lentos é lenta por uma de quatro razões: leem demais, recomputam o mesmo trabalho, trazem dados para o driver ou fazem fan-out sem materializar. Cada uma tem uma correção direta na linguagem.

Leia menos

A linha mais barata é a que nunca sai da fonte.

// tabela de lake particionada: filtre pela coluna de partição primeiro
{ "type": "filter", "condition": "ordered_at >= '{since}'" }

// depois estreite as colunas, antes de qualquer join
{ "type": "select", "columns": ["id", "customer_id", "amount"] }

A ordem importa: um select depois do join ainda carrega todas as colunas pelo shuffle.

Faça disso um hábito: comece toda cadeia com filter e depois select, antes de qualquer join, struct, group_by ou window. O otimizador do Spark empurra parte disso sozinho, mas fazer explícito encolhe os dados para todo passo seguinte, ajuda o planner e faz o pipeline ler-se como “estreite cedo, trabalhe depois”.

Reutilize a entrada sem reler

Quando você precisa fazer self-join da entrada — ou rodar SQL sobre ela — sem ler a fonte duas vezes, registre-a uma vez como temp view cacheada com input_view:

fw.run("orders.json", input_view="orders")            # temp view de sessão, cacheada
fw.run("orders.json", input_view={"name": "orders", "type": "global"})  # global_temp.orders

// um passo seguinte junta a entrada com ela mesma pela view — sem segunda leitura
{ "type": "join",
  "with": { "format": "view", "path": "orders" },
  "with_transformations": [
    { "type": "group_by", "by": ["customer_id"], "agg": ["max(amount) as top"] }
  ],
  "on": "customer_id", "how": "left" }

O DataFrame de entrada é cache()ado e registrado antes das transformações rodarem, então o self-join reutiliza as linhas em cache em vez de recomputar a linhagem da fonte. Defina no construtor Sparquet(...) para valer em toda execução, ou por chamada de run(...).

Faça checkpoint do plano

{ "type": "checkpoint", "method": "localCheckpoint", "eager": true }

Três lugares justificam:

Depois de joins pesados, para o plano não crescer sem parar.
Antes de collect, uma action no driver que senão recomputaria a linhagem inteira.
Antes de espalhar para vários destinos, para o trabalho acontecer uma vez em vez de uma por destino.

Empurre a lista de chaves

Quando o conjunto de trabalho é pequeno e a tabela que você enriquece é enorme:

[
  { "type": "checkpoint" },
  { "type": "collect", "column": "customer_id", "as": "customers" },
  {
    "type": "join",
    "with": { "format": "delta", "path": "sales.bronze_events" },
    "with_transformations": [
      { "type": "filter", "condition": "customer_id IN ({{customers}})" }
    ],
    "on": "customer_id"
  }
]

O IN (...) literal deixa Delta e Parquet pularem arquivos usando suas estatísticas. Mantenha a lista em dezenas de milhares de chaves no máximo — além disso, um join comum é a ferramenta certa. Veja Joins e pushdown.

Prove que o pushdown aconteceu

{ "type": "debug", "label": "leitura", "actions": ["pushdown"] }

O explain sugere; isto responde. Por nó de leitura, imprime PartitionFilters, PushedFilters, PushedAggregates, PushedGroupBy, RuntimeFilters e quantas colunas o scan devolve, e conta os nós Filter acima dos scans — predicado avaliado depois da leitura é dado que subiu do disco para ser descartado. O que isso torna visível:

um filtro em coluna de partição cai em PartitionFilters e pula diretórios inteiros; um filtro em coluna comum cai em PushedFilters e só pula os row groups que as estatísticas de min/max descartam.
um predicado que a fonte não sabe expressar — UDF, cast, lower(col) = 'x' — permanece como Filter acima do scan. Reescreva-o com a coluna nua de um dos lados.
em JDBC, só a leitura de tabela empurra. Com query, o SELECT que você escreveu já é o corte.
RuntimeFilters é dynamic partition pruning: a tabela fato podada pelo filtro da dimensão, decidido em runtime.

Não custa nada — ler o plano físico é planejamento, não execução.

Alavancas de arquivo

Os defaults servem para a maioria dos pipelines. Estas são as que valem conhecer quando não servem; defina-as em spark.configs.

Config	Default	O que faz
`spark.sql.parquet.filterPushdown`	`true`	empurra o predicado para as estatísticas de row group do Parquet
`spark.sql.parquet.aggregatePushdown`	`false`	responde `min`/`max`/`count` pelo footer, sem ler dados
`spark.sql.orc.filterPushdown` / `.aggregatePushdown`	`true` / `false`	o mesmo, para ORC
`spark.sql.optimizer.dynamicPartitionPruning.enabled`	`true`	poda a tabela fato com o filtro da dimensão em runtime
`spark.sql.optimizer.runtime.bloomFilter.enabled`	`true`	bloom filter na chave do join, corta o que vai para o shuffle
`spark.sql.files.maxPartitionBytes`	`128MB`	tamanho da task na leitura de arquivos
`spark.sql.adaptive.coalescePartitions.enabled`	`true`	funde partições minúsculas depois do shuffle

O aggregatePushdown é o que vale testar à mão: com ele ligado, um count(1) ou um max(dt) sobre Parquet é respondido pelos footers dos arquivos e não lê dado nenhum — mas vem desligado porque só se aplica a uma agregação nua sobre o scan, e qualquer filtro ou cast no meio devolve o ganho em silêncio.

Opções de leitura da mesma família: mergeSchema (desligado por default, e cada arquivo extra lido custa uma listagem), basePath (de qual prefixo as colunas de partição são derivadas), recursiveFileLookup e pathGlobFilter (ler menos por não listar), e spark.sql.sources.partitionOverwriteMode: dynamic, que faz o overwrite substituir apenas as partições presentes no DataFrame em vez do diretório inteiro.

Pare cedo

{ "type": "stop_if_empty", "message": "Nada a processar" }

Colocado logo após o filtro que define o conjunto de trabalho, transforma uma noite de no-op em segundos em vez de um pipeline inteiro sobre zero linhas.

Modele o lado direito de um join

"with_transformations": [
  { "type": "filter", "condition": "active = true" },
  { "type": "select", "columns": ["customer_id", "segment"] },
  { "type": "distinct" }
]

Filtre, projete, deduplique — nessa ordem — antes do join. Um lado direito que repete a chave multiplica linhas, o que é bug de correção e de performance ao mesmo tempo.

Skew são dois problemas diferentes

Skew de join, na leitura — uma chave com muito mais linhas que as outras deixa uma task só rodando por minutos enquanto as demais já terminaram. Este o AQE resolve: spark.sql.adaptive.skewJoin.enabled (default true), com skewedPartitionFactor (5) e skewedPartitionThresholdInBytes (256MB) definindo o que conta como skew; ele divide a partição problemática e replica o lado correspondente. Só age em sort-merge join com o AQE ligado, então um join com hint de broadcast nunca chega lá.
Skew de escrita — um valor da coluna de partition_by que concentra a maior parte das linhas grava um arquivo enorme enquanto os outros diretórios já acabaram. Aqui o AQE não ajuda, porque o layout de diretórios é o requisito, não um acidente do plano. Divida de propósito: repartition pelas colunas de partição mais um sal (pmod(hash(id), 8)), o que transforma o arquivo único daquele diretório em oito — e o leitor não paga nada pelos arquivos extras enquanto eles continuarem grandes.

Cuidado com a escrita

partition_by com alta cardinalidade cria milhares de arquivos minúsculos. Particione por algo grosso (data, país), nunca por um id.
merge custa mais que append. Use quando precisar de idempotência, não por padrão.
inferSchema do CSV custa um passo extra. Defina cast explícito e desligue-o para feeds grandes e estáveis.

DataFusion Comet, opt-in

Comet é um plugin do Spark que substitui operadores do plano físico por implementações nativas (Rust/Arrow) e cai de volta para o Spark no que não suporta. O framework não precisa de código para isso — é config de sessão:

Por que ligar. O benefício não é “Rust é rápido” — são quatro propriedades concretas:

Execução vetorizada sobre Arrow. O operador nativo trabalha em lotes colunares, com instruções SIMD, em vez de linha a linha na JVM. É por isso que o ganho aparece em varredura, filtro, projeção, hash aggregate e sort, e some onde o tempo é de rede.
Memória fora do heap. Os buffers de execução saem do heap da JVM, o que tira do coletor de lixo justamente quem mais aloca. Em job com pausa de GC visível na timeline do Spark, essa é a metade do ganho que se nota primeiro.
Nenhuma mudança no pipeline. É configuração de sessão, não API: o mesmo JSON, os mesmos readers, writers, transformações e validações. Nada no arquivo do pipeline diz que existe Comet.
Adoção parcial e reversível. A troca é operador a operador, e o que o Comet não suporta continua no Spark, dentro do mesmo plano. Não há migração nem porta de mão única: tirar as configs volta ao estado anterior na execução seguinte.

Compensa em leitura volumosa de Parquet com filtro, projeção, agregação, sort ou shuffle no caminho — o perfil da maior parte da ingestão em lote. Não compensa quando o tempo está do outro lado da rede (JDBC, APIs), com UDF em Python (que força o trecho de volta para o Spark), em pipeline dominado por escrita, nem em volume pequeno, onde dimensionar off-heap e carregar 88 MB de jar custa mais do que economiza.

{
  "spark": {
    "configs": {
      "spark.plugins": "org.apache.spark.CometPlugin",
      "spark.shuffle.manager": "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager",
      "spark.memory.offHeap.enabled": "true",
      "spark.memory.offHeap.size": "4g"
    }
  }
}

Medido no pipeline do próprio framework, 40.000.000 de linhas / 290 MB de Parquet, local[4], três repetições cronometradas depois de um aquecimento descartado (mediana):

Forma	Sem Comet	Com Comet	Ganho
filtro + `group_by` com `sum`/`count`	3,56s	1,61s	2,21x
filtro + `count`	1,02s	0,83s	1,24x

O ganho acompanha quanto do plano virou nativo: a agregação voltou com CometNativeScan, CometFilter, CometHashAggregate, CometExchange e CometNativeShuffle, enquanto a forma de contagem tem muito menos a acelerar. Antes de ligar:

o jar precisa estar no classpath do driver antes de a JVM subir — --driver-class-path, spark.driver.extraClassPath ou o diretório de jars do cluster. Por spark.jars ele chega tarde e a sessão morre com java.lang.ClassNotFoundException: org.apache.spark.CometPlugin.
o artefato é por linha do Spark: org.apache.datafusion:comet-spark-spark4.1_2.13:1.0.0 para o Spark 4.1, cerca de 88 MB.
a lib nativa existe só para Linux. Em qualquer outro sistema o plugin carrega, se desabilita em silêncio e o pipeline passa sem aceleração nenhuma — então confira os nós Comet* no plano em vez de presumir.
memória off-heap é obrigatória (spark.memory.offHeap.enabled e .size).
o fallback é por operador, e spark.comet.explain.fallback.enabled registra o motivo de cada um.

Meça antes de adivinhar

{ "type": "debug", "label": "after join", "actions": ["count", "explain"], "extended": false }

explain mostra o plano que o Spark vai realmente rodar — o jeito mais rápido de confirmar que um filtro foi empurrado ou que um broadcast aconteceu. count após cada estágio diz onde as linhas se multiplicaram.

O painel Run no Studio reporta linhas lidas, linhas escritas e duração por execução, o que costuma bastar para identificar a regressão sem abrir a Spark UI.

Um checklist rápido

Sintoma	Causa provável
Tempo escala com o número de saídas	Falta `checkpoint` antes do fan-out
Driver sem memória	`collect` numa coluna grande, ou uma lista `IN (...)` enorme
Join demora mais que todo o resto	Lado direito não filtrado/projetado antes do join
Milhares de arquivos minúsculos	`partition_by` numa coluna de alta cardinalidade
Lento toda noite, mesmo sem dados	Falta `stop_if_empty`
Um `Filter` aparece acima do scan no plano	Predicado que a fonte não expressa — confirme com `debug` / `pushdown`
Uma task roda por minutos enquanto as outras já acabaram	Skew de join (AQE), ou um valor de partição concentrando as linhas (escrita)