o problema
O shuffle do Spark grava em disco, ordenado pelo hash da chave de partição. Cada map task produz dois arquivos: um com os dados, agrupados em blocos por partição, e um índice com os offsets. Quando o reduce começa, ele abre conexão com cada Spark ESS que guarda parte da sua entrada e pede um bloco por request. O ESS faz seek, lê aquele bloco e devolve. Na LinkedIn isso acontece em escala de petabytes por dia e dezenas de bilhões de blocos por dia, sobre HDD. O bloco médio tem dezenas de KB. É leitura aleatória pequena, lida uma única vez e em ordem aleatória — cache não ajuda —, então o disco bate no teto de IOPS e o fetch atrasa. Os autores medem o preço: em torno de 15% de todo o recurso de computação do Spark do cluster se perde nessa espera. E há o lado da confiabilidade: para S shuffle services e E executors são até S*E conexões, com S e E chegando a 1.000 cada; uma falha de conexão mata o reduce stage inteiro e obriga a recomputar o stage anterior.
O reflexo é mexer no número de mappers e reducers. Não resolve. Existe o “tuning dilemma”: mudar os parâmetros para aumentar o bloco significa processar mais dado por tarefa, o que piora outra coisa. Pior, a prática que funciona quando o workload cresce é manter o dado por tarefa constante e escalar horizontalmente — e aí M*R cresce quadraticamente com o volume, então o bloco médio encolhe por construção. Trocar HDD por SSD tampouco serve: caro nessa escala e o SSD se desgasta rápido guardando dado temporário.
a ideia
Inverter a direção da transferência. Em vez de o reduce sair puxando milhares de fragmentos espalhados por mil máquinas, cada map task, depois de já ter materializado seu arquivo de shuffle normalmente, empurra os blocos para shuffle services remotos escolhidos pelo driver. Todos os blocos da mesma partição, venham de que mapper vierem, vão para o mesmo serviço, que faz append deles num único arquivo mesclado por partição. O reduce depois lê esse arquivo grande, quase sempre local à própria tarefa.
O detalhe que sustenta o desenho é o merge ser opcional. O push é uma segunda cópia; o shuffle original continua lá. Se o push falhar, se der colisão no merge, se o driver mandar parar antes da hora, o reduce busca os blocos originais e ninguém percebe.
como funciona
Terminado o arquivo de shuffle, o mapper divide os blocos em chunks contíguos de alguns MB, cada chunk endereçado a um shuffle service segundo um mapeamento que todos os mappers calculam igual (Algoritmo 1). A ordem dos chunks é randomizada, para reduzir a chance de dois mappers empurrarem a mesma partição ao mesmo tempo. O envio sai do caminho crítico: vai para um thread pool dedicado, o map task termina antes.
No serviço, cada partição ativa tem um arquivo mesclado e um bloco de metadados num ConcurrentHashMap, chaveado por app ID, shuffle ID e partição: um bitmap com os mapper IDs já mesclados, um offset da última escrita bem-sucedida e o currentMapId em andamento. O bitmap descarta duplicata, o currentMapId serializa escritas concorrentes na mesma partição, e o offset conserta corrupção — o próximo bloco sobrescreve o pedaço quebrado, e se o quebrado era o último ele é truncado na finalização. Nada é ordenado nem bufferizado em memória no serviço: ele só faz append.
O driver junta M MapStatus e R MergeStatus e sabe exatamente o que está mesclado, onde, e quais blocos originais respaldam cada arquivo mesclado. Para não perder paralelismo entre fetch e execução, o arquivo mesclado é fatiado em slices de MB registradas num índice próprio, e o reduce busca as slices em vez de um bloco único gigante.
o que isso custou
O shuffle é escrito duas vezes. Os autores defendem a troca porque escrita aleatória pequena é agrupada por page cache e buffer de disco, coisa que leitura aleatória pequena não tem — e o benchmark confirma: transferir 150 GB em blocos de 10 KB por um único serviço leva quatro horas no Spark original e pouco mais de cinco minutos no push do Magnet. O armazenamento também dobra, o que passa porque o dado é temporário: o pico é de centenas de TB.
O resto são recuos assumidos. Straggler no push atrasa direto o job, então o driver impõe um teto de espera e manda parar o merge — alguns blocos ficam sem mesclar. Partições enviesadas acima de um limiar são puladas de propósito, e ficam por conta do adaptive execution. O job pequeno não ganha nada, e o map stage às vezes fica mais longo (7,1 para 8,5 minutos no teste combinado). E há o custo de arquitetura: por integrar-se ao shuffle nativo do Spark, o Magnet é acoplado ao Spark, enquanto Sailfish e Cosco, que delegam o shuffle a um armazenamento externo, se estendem mais fácil a outros engines.
onde isso aparece hoje
Nos números de produção da própria LinkedIn: quase 30% de queda no runtime end-to-end, com o reduce stage do workload I/O-intensivo caindo 81% e o tempo total de tarefa caindo 66%. A implementação foi feita sobre o Spark 2.3 sem tocar em API pública, e o push-based shuffle acabou incorporado ao Apache Spark upstream, ligável por configuração.
O problema atacado aqui é herança direta do modelo de MapReduce: o all-to-all entre map e reduce é a parte que não escala de graça. E é sobre a materialização em disco introduzida pelos RDDs que o Magnet trabalha — ele não substitui esse mecanismo, empilha uma segunda cópia mesclada em cima dele.