o problema
Entre 1999 e 2004 o Google escreveu centenas de computações de propósito específico sobre dados brutos: documentos rastreados, logs de requisição. O que sai delas é simples de descrever — índice invertido, grafo de links da web, contagem de páginas por host, as consultas mais frequentes do dia. Conceitualmente, quase tudo é uma passada sobre registros seguida de uma agregação por chave.
O problema é que a entrada não cabe numa máquina. Para terminar em tempo razoável, a computação tem que rodar em centenas ou milhares de máquinas, e aí o código de dez linhas que descreve a intenção some no meio do código que particiona a entrada, distribui os pedaços, detecta que a máquina 417 morreu, refaz o trabalho dela, e equilibra a carga entre as que sobraram. Cada equipe reescrevia essa mesma parte, sempre de novo, sempre um pouco diferente. Em máquinas commodity a falha não é evento raro: nos jobs de agosto de 2004, a média foi de 1,2 mortes de worker por job.
a ideia
O usuário escreve duas funções. map recebe um par chave/valor e emite um conjunto de pares intermediários. reduce recebe uma chave intermediária e um iterador sobre todos os valores associados a ela, e devolve um conjunto menor — tipicamente zero ou um valor. Nada mais. Sem sockets, sem barreira, sem retry.
O ganho não vem de map e reduce serem primitivas especialmente expressivas. Vem do que elas proíbem. Como cada invocação de map é função dos seus próprios dados e de mais nada, a biblioteca pode decidir sozinha em que máquina rodar cada pedaço, rodar o mesmo pedaço duas vezes em máquinas diferentes, ou jogar fora um resultado e refazê-lo. A tolerância a falha inteira é re-execução, e re-execução só é barata porque o modelo é restrito.
como funciona
A biblioteca quebra a entrada em M splits de 16 MB a 64 MB e o espaço de chaves intermediárias em R partições, por padrão hash(key) mod R. Uma cópia do programa vira master; as outras viram workers. O master mantém, para cada uma das M+R tarefas, o estado (idle, in-progress, completed) e a máquina responsável.
O map worker lê seu split, chama a função do usuário e acumula os pares em memória. Periodicamente esse buffer vai para o disco local, já separado nas R regiões; as localizações voltam para o master, que as empurra incrementalmente para os reduce workers. O reduce worker puxa esses pedaços por RPC dos discos locais dos map workers, ordena tudo por chave intermediária — external sort se não couber na memória —, itera sobre o resultado ordenado e escreve num arquivo de saída por partição. Ninguém junta os R arquivos no fim: eles costumam virar entrada do próximo MapReduce.
Três detalhes carregam o desempenho. O master consulta o GFS e tenta agendar cada map task na máquina que já tem uma réplica do bloco, ou pelo menos no mesmo switch. O commit é atômico: tarefa em andamento escreve em arquivo temporário e o reduce faz rename no fim, então execução duplicada não corrompe a saída. E quando o job está perto de acabar, o master dispara backup executions das tarefas que ainda não terminaram — vale a primeira que completar. A escala típica: M = 200.000, R = 5.000, 2.000 workers.
o que isso custou
O master é ponto único de falha, e a implementação descrita não faz failover: se ele morre, o job aborta e o cliente que tente de novo. Os autores dizem que checkpoint seria fácil, e simplesmente não fizeram, porque é uma máquina só contra milhares.
O estado no master é O(M * R), o que limita quão fino dá para picar o trabalho, e R fica amarrado ao número de arquivos de saída que o usuário aguenta. Skipping bad records é assumidamente uma gambiarra: um signal handler que dispara um pacote UDP de “last gasp” com o número de sequência do registro antes de o processo morrer. Não há commit de duas fases para múltiplos arquivos de saída, então efeito colateral atômico e idempotente é responsabilidade de quem escreve o job. E o benchmark de sort usava uma partitioning function com conhecimento embutido da distribuição das chaves — os próprios autores notam que um sort de verdade precisaria de uma passada anterior de amostragem.
Há um custo estrutural que o texto não chama de custo: o intermediário vai para disco e o reduce só começa de fato depois que os maps terminam. O sort de 1 TB levou 891 segundos, contra 1.057 do melhor TeraSort reportado na época — mas escreveu 2 TB de saída, porque replicação era o mecanismo de confiabilidade disponível. Os autores observam que erasure coding no sistema de arquivos reduziria essa banda.
onde isso aparece hoje
Dentro do Google, o efeito imediato foi a reescrita do sistema de indexação da busca: mais de 20 TB de documentos brutos passando por uma sequência de cinco a dez operações MapReduce, com uma fase caindo de cerca de 3.800 linhas de C++ para cerca de 700. Em setembro de 2004 havia quase 900 programas MapReduce distintos no repositório.
Fora dele, este paper e o do GFS foram a especificação a partir da qual o Hadoop foi construído, e com ele a década de processamento em lote sobre HDFS. A geração seguinte de motores de dados nasceu atacando exatamente o ponto caro descrito aqui — materializar o intermediário em disco a cada estágio, o que dói em carga iterativa. E a ideia de backup task, disparar uma cópia redundante para não ficar refém do pior worker, saiu do batch e virou prática corrente de controle de latência de cauda em sistemas online.