Processamento de Dados Massivos/O ambiente Hadoop

Hadoop

Apesar da importância conceitual dos trabalhos sobre o Google File System e o modelo MapReduce, o sistema descrito pelos autores se manteve propriedade única da Google, não tendo sido distribuído externamente. Isso mudou apenas com a divulgação do Hadoop, um ambiente de código aberto que implementa os princípios daqueles dois sistemas de forma bastante aproximada, apesar de algumas diferenças. Hadoop foi desenvolvido originalmente por Michael J. Cafarella e Doug Cutting, este último funcionário da Yahoo!. Com o passar do tempo, o projeto passou a ser hospedado pela Apache Software Foundation e vem recebendo contribuições de diversas fontes, principalmente da própria Yahoo! .

As principais diferenças entre Hadoop e o modelo original da Google são a linguagem (Hadoop é escrito em Java, enquanto GFS/MR são escritos em C/C++) e o fato do HDFS não permitir a operação de escrita ao fim de arquivos já criados (append). No HDFS, arquivos só podem ser abertos para escrita quando de sua criação; depois disso se tornam apenas de leitura. Essas duas diferenças acarretam algumas outras, como a mudança da interface de programação, uma maior amarração entre a interface do sistema de arquivos HDFS e classes Java para o processamento dos conteúdo dos arquivos em função do tipo do conteúdo e uma extensão do modelo de entrada para lidar com diretórios (usar todos os arquivos em um diretório como entrada) ao invés de apenas arquivos isolados.

Esta última alteração se deve a uma simplificação do HDFS relacionada à forma de atualização de arquivos. Normalmente, ao final de uma aplicação MapReduce, cada nó que executa uma tarefa de redução gera um arquivo de saída independente, numerado em função dos nós participantes da aplicação. Isso porque o HDFS não permite que diversos processos escrevam em um mesmo arquivo em paralelo. Já que a saída das aplicações gera arquivos dessa forma, a entrada do programa também aceita a indicação de um diretório, de forma que a saída de uma aplicação possa ser usada como entrada para outras.

Um exemplo de programação

Considerando-se as diferenças de implementação, no caso do Hadoop vale a pena analisar detalhadamente o código realmente utilizado para a aplicação ContaPalavras, ao invés de apenas um pseudo-código mais abstrado. O código apresentado a seguir foi obtido do tutorial sobre Hadoop da Yahoo! , uma fonte de consulta altamente recomendada para detalhes sobre elementos de instalação, programação e execução no Hadoop. Outra fonte que merece menção é o tutorial da Apache Software Foundation .

A função map define os tipos dos pares chave/valor de entrada e saída (linha 2). Neste caso, para um arquivo texto, Hadoop entrega cada linha com uma chave que corresponde à posição (em bytes) do começo da linha (texto) em relação ao começo do arquivo, daí usar um inteiro longo. A chave produzida pelo map será uma palavra (texto) e o valor será um inteiro. Os tipos terminados em Writable são usados pelo Hadoop para lidar com a leitura e escrita dos mesmos nos arquivos do HDFS corretamente (serialização e desserialização).

Outros detalhes de configuração são a definição do tipo de arquivo de saída (OutputCollector) que receberá os pares chave/valor do tipo indicado (linha 8). A classe Reporter é usada para gerar dados de acompanhamento da execução para o sistema.

O código realmente responsável pelo processamento está entre as linhas 10 e 15: o valor recebido como entrada é transformado em um string, que é dividido em tokens, que são enviados para a saída com a chamada de output.collect.

O código do redutor, apresentado a seguir, também exige detalhes de declaração de tipos de entrada e saída, que neste caso são iguais: palavra(texto)/inteiro (linha 2). Como a função de redução recebe uma lista de valores, o segundo tipo é um iterador sobre a lista de valores associados a cada chave (linha 4). Os outros elementos da definição são semelhantes ao caso do map.

O núcleo de processamento está entre as linhas 7 e 11 nesse caso: o loop itera sobre todos os valores recebidos para cada palavra (todos valendo um, nesse caso, já que não foi definido um combiner) e o valor final da contagem para cada palavra é lançado com um output.collect.

Finalmente, Hadoop requer um código de configuração que identifique para o sistema as funções a serem usadas, os arquivos de entrada e saída com seus tipos, etc. Essas informações são definidas por um objeto de configuração que é construído para esse fim como na listagem a seguir.

As linhas 8 e 9 identificam as classes que definem as operações de mapeamento e redução. As linhas 11 e 12 associam arquivos armazenados no HDFS à entrada e saída do programa. O nome usado como entrada pode ser o de um arquivo específico ou de um diretório contendo diversos arquivos, que serão todos processados da mesma forma, potencialemente em paralelo. Já o nome usado para a saída deve sempre ser o de um diretório vazio, onde serão criados arquivos para cada nó de processamento que executará a função de redução.

O tipo dos pares chave/valor de saída é definido pelos comandos nas linhas 5 e 6. Com os comandos usados, a saída será gerada como arquivo texto, com um caractere de tabulação separando a chave do valor em cada linha. Como não há comandos semelhantes para a entrada, o tipo assumido é o de arquivo de texto simples. O sistema oferece também a opção de tratar a entrada como um arquivo texto com chaves e valores separados por um caractere de tabulação (como o gerado pela saída do programa), ou como um arquivo com registros complexos, caso em que o programador deverá fornecer uma classe de leitura adequada. Neste último caso há detalhes que precisam ser tratados em relação ao processamento de registros que ultrapassam o limite de um bloco do arquivo no HDFS. Este e outros detalhes da interface de programação podem ser encontrados no tutorial da Yahoo! .

Ao se executar o método runJob do objeto de configuração que descreve a aplicação (linha 14), o ambiente de tempo de execução do Hadoop passa a executar a tarefa. Os passos envolvidos nesse processo são descritos a seguir.

Ambiente de execução

Para melhor compreender o comportamento de uma aplicação paralela Hadoop é importante entender os passos de sua execução. Esses passos são ilustrados na figura abaixo e descritos a seguir. Os números entre parênteses se referem aos passos na figura.

Representação do ambiente de execução do Hadoop.

Ao disparar o programa da aplicação, o Hadoop dispara os processos que farão parte da execução (1). Um processo mestre, responsável por coordenar os demais, é disparado junto à aplicação original do usuário e processos trabalhadores são disparados no conjunto de máquinas configurados durante a instalação do Hadoop.

Com base na informação sobre o arquivo de entrada, o mestre identifica os trabalhadores que foram disparados mais próximo de cópias dos blocos do arquivo e atribui a cada um deles parte das tarefas de map, identificando os pedaços do arquivo que devem ser processados por cada trabalhador, denominados splits (2). Os trabalhadores começam então a ler pedaços do arquivo e produzir os pares chave/valor para a tarefas de map (3). Os pares chave/valor produzidos nessa etapa são assinalados para um dos processos trabalhadores escolhidos para a redução por uma função de assinalamento. Essa função é normalmente uma função de hash simples, parametrizada pelo número de redutores, mas pode ser substituída por outro tipo de mapeamento definido pelo usuário. O resultado do map é então armazenado localmente na máquina em que cada trabalhador executa, agrupado pelos identificadores dos processo de redução assinalados pela função de mapeamento (4). À medida que tarefas de mapeamento sobre splits completam, os trabalhadores informam o mestre sobre quais chaves foram geradas no processo e identificam os arquivos gerados para cada nó responsável pela redução (5).

O mestre, à medida que cada split do arquivo de entrada é completado, coleta a informação sobre os arquivos intermediários e notifica cada nó trabalhador que executará um reduce sobre quais nós de map já têm arquivos intermediários com chaves destinadas a eles (6). Os redutores então contactam os nós identificados e vão requisitando esses arquivos usando consultas HTTP. Os pares chave/valor recebidos vão sendo ordenados por suas chaves, para garantir a geração correta das listas de valor para cada chave de forma completa (7). Quando os trabalhadores que executam as tarefas de mapeamento terminam e todos os pares chave/valor intermediários estão disponíveis para os processos de redução adequados, a função reduce começa a ser executada para cada chave assinalada para um trabalhador pela função de mapeamento. A saída de cada trabalhador envolvido na redução é escrita em um arquivo individual no HDFS, todos no mesmo diretório indicado pelo usuário na configuração da aplicação (8).

O número de trabalhadores pode ser configurado pelo usuário, mas em geral é calculado automaticamente pelo Hadoop, em função do tamanho da entrada (número de splits) e do número de máquinas disponíveis.

Como no MapReduce original, Hadoop usa replicação de tarefas para conseguir balanceamento de carga e tolerância a falhas. À medida que as tarefas de um certo tipo vão terminando, o processo mestre dispara novas execuções de tarefas que estão levando mais tempo para serem completadas em outros nós. Além disso, ele monitora o estado de todos os trabalhadores a fim de detectar nós que tenham deixado de responder. Nesse caso, ele identifica todas as tarefas que haviam sido atribuídas a um nó (sejam maps ou reduces) e as atribui novamente a outros nós ainda ativos.