Partições, repartition e coalesce

Um DataFrame é dividido em partições, e cada partição é processada por uma tarefa num núcleo. Com 8 núcleos e 200 partições, o estágio roda em ondas de até 8 tarefas por vez.

De onde vem o número de partições

• Na leitura de arquivos: pelo tamanho, com spark.sql.files.maxPartitionBytes (padrão 128 MB) por partição, aproximadamente.
• Depois de um shuffle (groupBy, join, distinct, orderBy): spark.sql.shuffle.partitions (padrão 200). Com a execução adaptativa (AQE, ligada por padrão desde o Spark 3.2), partições pequenas são juntadas depois do shuffle.

df = spark.read.parquet("/mnt/raw/pedidos")
print(df.rdd.getNumPartitions())
print(spark.conf.get("spark.sql.shuffle.partitions"))

16
200

(O primeiro número depende do tamanho dos arquivos e do número de núcleos: o exemplo supõe uns 2 GB num cluster pequeno, 2 GB ÷ 128 MB. O segundo é o padrão.)

repartition: shuffle completo, partições equilibradas

repartition(n) redistribui todas as linhas pela rede em n partições do mesmo tamanho (ou por coluna, repartition(n, "data"), juntando as linhas de mesma chave). Serve para aumentar o paralelismo ou preparar uma escrita particionada.

coalesce: sem shuffle, só junta

coalesce(n) só diminui: junta partições vizinhas no mesmo executor, sem mover dado pela rede. É barato, mas herda o desequilíbrio: se uma partição era enorme, continua enorme.

grande = df.repartition(64)       # shuffle: 64 partições parecidas
print(grande.rdd.getNumPartitions())
poucas = grande.coalesce(8)       # sem shuffle: 8 partições
print(poucas.rdd.getNumPartitions())
print(poucas.coalesce(100).rdd.getNumPartitions())  # coalesce não aumenta

64
8
8

Skew: uma partição que demora por todas

O estágio só termina quando a última tarefa termina. Se uma chave concentra metade das linhas, a tarefa dela vira o gargalo, e o resto do cluster espera parado. Sinais: no Spark UI, uma tarefa com duração muito acima da mediana.

• A AQE trata skew em join (spark.sql.adaptive.skewJoin.enabled), quebrando a partição grande.
• Sem AQE, a saída clássica é o *salting*: acrescentar um sufixo aleatório à chave quente para espalhá-la.
• coalesce(1) antes de gravar um arquivo único põe todo o trabalho anterior numa tarefa só, se o Spark conseguir empurrá-lo para dentro do mesmo estágio. Para um arquivo só, prefira repartition(1).