Ajuste de Shuffle do Spark no EMR Serverless para Grandes Volumes

Fazer o ajuste de shuffle do Spark no EMR Serverless reduz o tempo de execução e o custo de storage.

Ao executar transformações amplas como junções (joins), agregações e funções de janela em grandes data lakes na nuvem, o Apache Spark transfere pares de chave-valor intermediários pela rede. Essa fase, conhecida como shuffle, representa o gargalo operacional mais comum em sistemas de produção. Em ambientes de computação sem servidor (serverless), onde os nós de processamento físico são abstraídos, operações de shuffle não otimizadas se traduzem diretamente em custos elevados e falhas frequentes por estouro de memória (Out-Of-Memory - OOM). Otimizar esse padrão de movimentação de dados é fundamental para alcançar eficiência financeira e previsibilidade de execução.

Historicamente, os clusters Hadoop locais dependiam de configurações do HDFS local ou SSDs físicos dedicados para gravar arquivos temporários de shuffle. Em uma arquitetura serverless como o Amazon EMR Serverless, o tamanho do disco local é desacoplado do perfil da instância e faturado dinamicamente com base na alocação. Compreender como alinhar a alocação de memória com a taxa de transferência do disco e o uso da rede é crítico ao gerenciar plataformas de análise que processam volumes de múltiplos terabytes.

A Anatomia do Shuffle no Spark em Ambientes Serverless

Para otimizar a fase de shuffle, é necessário rastrear o caminho dos dados desde o estágio de mapeamento (map) até o estágio de redução (reduce). Quando um shuffle é disparado, as tarefas de mapeamento (MapTasks) gravam arquivos de segmento de partição no disco local do container de execução. Posteriormente, as tarefas de redução (ReduceTasks) buscam esses arquivos através da rede a partir de outros nós do cluster. No EMR Serverless, os dados do shuffle são gravados no armazenamento efêmero alocado para cada worker ativo.

Por padrão, o EMR Serverless provisiona os nós de execução com 20 GB de capacidade de disco local. Se os seus conjuntos de dados intermediários excederem esse limite, o motor do Spark gera exceções de gravação em disco, causando a falha do job de processamento. Embora seja possível expandir o armazenamento em disco para até 200 GB por worker, aumentar esses limites sem ajustar os parâmetros de software resulta em superprovisionamento de recursos e elevação dos custos operacionais. O objetivo deve ser maximizar o rendimento da memória, utilizar serialização eficiente e minimizar o volume de dados gravados fisicamente em disco.

Adicionalmente, os nós de execução serverless não compartilham um daemon de disco local persistente, como o serviço de shuffle externo em ambientes YARN tradicionais. Em vez disso, eles dependem do BlockManager do Spark para transferir blocos diretamente entre executores ativos. Se um executor for encerrado devido à inatividade pelo auto-escalador serverless, os blocos de shuffle armazenados nele podem ser perdidos, forçando o Spark a recalcular as etapas anteriores da pipeline. Equilibrar a agressividade do escalonamento automático com a distribuição de tarefas é crucial para jobs que lidam com dados altamente desbalanceados.

Estratégias de Ajuste de Memória e Disco

Para equilibrar as alocações de memória e disco temporário local, os engenheiros precisam ajustar os perfis de execução. A memória total do executor é dividida entre memória de armazenamento (utilizada para cache) e memória de execução (utilizada para shuffles e joins). Ao ajustar essas proporções, você mantém mais dados intermediários na RAM, evitando a degradação de performance associada à gravação em discos SSD locais.

# Configuração recomendada de Spark para otimização no EMR Serverless
from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("EMR-Serverless-Shuffle-Optimization")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.kryoserializer.buffer.max", "1024m")
    .config("spark.sql.shuffle.partitions", "2000")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", "134217728")
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .config("spark.shuffle.file.buffer", "1m")
    .config("spark.reducer.maxSizeInFlight", "128m")
    .config("spark.shuffle.service.enabled", "false")
    .config("spark.dynamicAllocation.enabled", "true")
    .getOrCreate())

A configuração apresentada prioriza a eficiência de memória durante o ciclo de vida do shuffle. Primeiramente, alterar o serializador padrão do Java para o KryoSerializer reduz o tamanho dos objetos na memória e os volumes transmitidos na rede, diminuindo o overhead de serialização de 40% a 60%. Elevar o parâmetro spark.shuffle.file.buffer do padrão de 32 KB para 1 MB permite que o Spark acumule blocos maiores na memória antes de gravá-los no disco físico, reduzindo as operações de E/S (I/O).

Da mesma forma, o aumento do parâmetro spark.reducer.maxSizeInFlight possibilita a busca de blocos maiores de dados de shuffle simultaneamente a partir de workers remotos. Isso reduz as idas e vindas de rede e mantém as CPUs dos workers ocupadas com o processamento de dados, evitando tempo ocioso durante etapas intensas de redução.

Execução Adaptativa de Consultas e Agrupamento de Partições

A Execução Adaptativa de Consultas (Adaptive Query Execution - AQE) é um mecanismo vital para otimizar os estágios de shuffle. Em operações tradicionais do Spark, a quantidade de partições de shuffle é predefinida pelo parâmetro estático spark.sql.shuffle.partitions (cujo padrão costuma ser 200). Se esse número for muito baixo para o seu volume de dados, as partições ficarão excessivamente grandes, resultando em problemas de falta de memória (OOM) e escrita em disco. Se for muito alto, gera-se uma sobrecarga de pequenas tarefas para o scheduler gerenciar.

O AQE ajusta dinamicamente a quantidade de partições em tempo de execução. Ao analisar as estatísticas das etapas concluídas após a fase de mapeamento, o AQE consegue agrupar partições pequenas ou dividir partições muito volumosas antes de iniciar a fase de redução.

Definir o parâmetro spark.sql.adaptive.advisoryPartitionSizeInBytes para 128 MB orienta o Spark a criar arquivos de partição que coincidam com os blocos recomendados de armazenamento. Para grandes conjuntos de dados, esse ajuste dinâmico garante que as tarefas utilizem a memória disponível dos workers serverless de forma equilibrada, sem causar falhas de execução.

Decisões Arquiteturais: Tabelas no S3 e Catálogos Externos

Ao desenhar pipelines na AWS, os engenheiros devem decidir onde persistir o estado intermediário. Embora o armazenamento local efêmero possua excelente desempenho, as arquiteturas serverless frequentemente se beneficiam ao salvar tabelas temporárias massivas diretamente no Amazon S3. Utilizar formatos de tabela modernos como Apache Iceberg ou Delta Lake fornece uma maneira estruturada de lidar com transformações pesadas de dados.

No nosso projeto de referência AWS Databricks Lakehouse, exploramos como os motores de processamento e armazenamento interagem quando desacoplados. Ao desenvolver pipelines complexos de múltiplas etapas, utilizar checkpoints físicos no S3 pode ser mais rápido e seguro do que depender de um único shuffle contínuo do Spark, especialmente quando o processamento consome várias horas.

Para shuffles excepcionalmente grandes, utilizar classes de armazenamento otimizadas ou integrações nativas do EMR Serverless ajuda a diminuir significativamente os tempos de processamento. Como detalhado no AWS Big Data Blog, configurar o armazenamento de maneira eficiente garante previsibilidade de custos mesmo sob demandas computacionais severas.

Implementando Ajustes de Shuffle em Produção

Para aplicar essas otimizações em seus fluxos de trabalho, adote um plano de execução estruturado:

1. Analise os logs existentes: Examine o Spark History Server e as métricas do cluster. Observe os valores de "Shuffle Spill (Disk)" e "Shuffle Spill (Memory)". Se a gravação em disco estiver alta, eleve a memória do executor ou configure discos maiores.
2. Identifique a assimetria dos dados: Se poucas tarefas consomem horas e as demais terminam rapidamente, há desbalanceamento de dados (skew). Ative o parâmetro spark.sql.adaptive.skewJoin.enabled para permitir que o Spark divida partições desbalanceadas dinamicamente.
3. Aplique as configurações: Insira os parâmetros customizados do Spark no JSON de submissão de tarefas do EMR Serverless, dentro da estrutura applicationConfiguration.
4. Monitore o impacto financeiro: Acompanhe o consumo nos painéis de faturamento da nuvem. Frequentemente, mitigar a gravação em disco gera uma redução direta no total de horas de computação faturadas.

Manter uma plataforma de dados confiável e de alta performance requer monitoramento contínuo. Centralizar e analisar métricas operacionais por meio de soluções como a nossa Data Observability Platform garante que as equipes de engenharia de dados detectem regressões de memória e desalinhamento de partições antes que afetem os sistemas de negócios.