Papers
MapReduce: Simplified Data Processing on Large Clusters
Jeffrey Dean, Sanjay Ghemawat · USENIX Symposium on Operating Systems Design and Implementation (OSDI) 2004, 2004
O problema: computações simples escondidas sob código de paralelização
Dean e Ghemawat partem de uma observação do dia a dia do Google: ao longo de cerca de cinco anos, eles e outros engenheiros escreveram centenas de programas especializados para processar grandes volumes de dados brutos, como documentos rastreados e logs de requisições, e gerar dados derivados, como índices invertidos, representações do grafo da web e contagens de páginas por host. Conceitualmente cada cálculo era simples. O problema é que os dados eram tão grandes que a execução precisava ser distribuída por centenas ou milhares de máquinas para terminar em tempo razoável. Cuidar de como paralelizar, distribuir os dados e lidar com falhas acabava obscurecendo a lógica original, que ficava enterrada em muito código complexo. A resposta dos autores foi criar uma abstração que permite expressar a computação desejada e esconde numa biblioteca os detalhes de paralelização, tolerância a falhas, distribuição de dados e balanceamento de carga. A inspiração vem das primitivas map e reduce de Lisp e de outras linguagens funcionais. A ideia é que a maioria das computações aplica uma operação map a cada registro lógico da entrada, gerando pares chave/valor intermediários, e depois uma operação reduce a todos os valores que compartilham a mesma chave. Com esse modelo, a re-execução vira o mecanismo principal de tolerância a falhas.
O modelo de programação: map, reduce e o exemplo de contagem de palavras
O usuário escreve apenas duas funções. Map recebe um par chave/valor de entrada e produz um conjunto de pares chave/valor intermediários. A biblioteca então agrupa todos os valores intermediários que têm a mesma chave intermediária e os entrega à função Reduce, que combina esses valores em um conjunto possivelmente menor, geralmente zero ou um valor de saída por chamada. Os valores chegam ao reduce por um iterador, o que permite tratar listas grandes demais para caber na memória. Em termos de tipos, map transforma (k1, v1) em uma lista de (k2, v2), e reduce transforma (k2, lista de v2) em uma lista de v2. O exemplo clássico é contar ocorrências de cada palavra numa grande coleção de documentos: o map percorre o conteúdo de cada documento e emite cada palavra associada ao valor 1, e o reduce soma todos os valores emitidos para uma mesma palavra. O paper mostra outros usos simples que cabem no mesmo molde: grep distribuído, contagem de frequência de acesso por URL, grafo reverso de links da web, vetor de termos por host, índice invertido e ordenação distribuída. Além das duas funções, o usuário preenche um objeto de especificação com nomes de arquivos de entrada e saída e parâmetros opcionais.
A execução: splits, master, arquivos intermediários e o papel do GFS
A implementação descrita foi feita para o ambiente do Google: grandes clusters de PCs comuns ligados por Ethernet, com discos IDE baratos, onde falhas de máquina são frequentes e os arquivos ficam num sistema de arquivos distribuído próprio (o GFS) que usa replicação. A biblioteca divide a entrada em M pedaços (splits), tipicamente de 16 a 64 MB, e inicia várias cópias do programa no cluster. Uma delas é o master, que atribui tarefas às demais, os workers, escolhendo workers ociosos para tarefas de map ou de reduce. Um worker de map lê seu split, aplica a função do usuário e guarda os pares intermediários em memória, gravando-os periodicamente no disco local, particionados em R regiões por uma função de particionamento, por padrão algo como hash(chave) mod R. O master repassa as localizações desses dados aos workers de reduce, que os leem remotamente por RPC dos discos das máquinas de map e os ordenam por chave, usando ordenação externa se não couberem na memória. Cada worker de reduce percorre os dados ordenados, chama a função do usuário para cada chave e anexa o resultado a um arquivo de saída. O resultado final são R arquivos de saída, um por tarefa de reduce. Para economizar largura de banda de rede, o master tenta escalar tarefas de map em máquinas que já guardam uma réplica do bloco de entrada, ou perto delas.
Tolerância a falhas, stragglers e tarefas de backup
O master verifica periodicamente cada worker. Se um worker não responde por um certo tempo, é marcado como falho. Tarefas de map já concluídas por ele voltam ao estado inicial e são reexecutadas, pois sua saída ficava no disco local da máquina agora inacessível. Tarefas de reduce concluídas não precisam ser refeitas, porque a saída delas já está no sistema de arquivos global. Tarefas em andamento no worker falho também voltam a ficar elegíveis para escalonamento. Num caso relatado, manutenção de rede derrubou grupos de 80 máquinas por vários minutos e o master simplesmente reexecutou o trabalho e concluiu a operação. A semântica se apoia em commits atômicos: cada tarefa grava em arquivos temporários, e o reduce renomeia atomicamente seu arquivo temporário para o final, de modo que, se o map e o reduce forem determinísticos, o resultado é igual ao de uma execução sequencial sem falhas. Como há um único master, a implementação descrita aborta a computação se ele falha, e o cliente pode tentar de novo. Outro problema é o straggler, uma máquina que demora muito para terminar uma das últimas tarefas. Perto do fim da operação, o master lança execuções de backup das tarefas restantes, e vale a primeira que terminar. Os autores dizem que isso costuma aumentar o uso de recursos em no máximo poucos por cento. No benchmark de sort, desligar esse mecanismo aumentou em 44% o tempo total.
Refinamentos, resultados medidos e uso real no Google
Além do núcleo, o paper descreve extensões úteis. Uma função combinadora faz um merge parcial dos dados em cada máquina de map antes do envio pela rede, o que ajuda quando há muita repetição de chaves, como em contagem de palavras. É possível definir uma função de particionamento própria, por exemplo para que todas as URLs de um mesmo host caiam no mesmo arquivo de saída. Dentro de cada partição, os pares são processados em ordem crescente de chave. Um modo opcional permite pular registros que causam falhas determinísticas, e há ainda execução local para depuração, páginas de status e contadores. Nos experimentos, num cluster de cerca de 1800 máquinas, um grep sobre 10^10 registros de 100 bytes terminou em cerca de 150 segundos, incluindo cerca de um minuto de overhead de inicialização. O sort do mesmo volume, cerca de 1 TB, levou 891 segundos, e com 200 processos worker mortos de propósito levou 933 segundos, apenas 5% a mais. Quanto ao uso, os autores contam que a primeira versão da biblioteca foi escrita em fevereiro de 2003 e que o número de programas MapReduce no código-fonte do Google cresceu de zero para quase 900 até setembro de 2004. Em agosto de 2004, uma amostra registrou 29.423 jobs, com tempo médio de 634 segundos. Os usos citados incluem aprendizado de máquina, agrupamento para o Google News e o Froogle, relatórios de consultas populares e computações em grafos.
Veja as ideias em ação
Pratique na escola de System Design os desenhos que usam este paper.