DataARTIGO

Processamento de Dados Distribuídos com Hadoop – Parte 01

Embora o Hadoop seja o núcleo da redução de dados para alguns dos
maiores mecanismos de pesquisa, ele é melhor descrito como uma estrutura
para o processamento de dados distribuídos. E não apenas dados, mas
grandes quantidades de dados, conforme necessário para mecanismos de
pesquisa e os dados de crawl que coletam. Como uma estrutura
distribuída, o Hadoop permite a operação de vários aplicativos que se
beneficiam com o paralelismo do processamento de dados.

As origens do Hadoop

O projeto Apache Hadoop foi inspirado e desenvolvido a partir de um
trabalho anterior do Google. Embora o Google tenha a patente desse
método de processamento de dados em larga escala, a empresa
generosamente concedeu uma licença para Hadoop.

A função deste artigo não é apresentar o Hadoop e sua arquitetura, mas
mostrar uma configuração de Hadoop simples. Através da seção Recursos,
é possível encontrar mais detalhes sobre a arquitetura, os componentes e
a teoria de operação do Hadoop. Com a renúncia de responsabilidade em
vigor, vamos direto à instalação e à configuração do Hadoop.

Configuração Inicial

Para esta demonstração, usaremos a distribuição Cloudera Hadoop. Você
encontrará nela suporte para uma variedade de distribuições Linux
diferentes, portanto, é ideal para começar.

Este artigo
considera primeiro que o sistema tenha a tecnologia Java (no mínimo o
release 1.6) e cURL instalado. Caso contrário, será preciso adicioná-la
primeiro (consulte a seção Recursos para informações mais detalhadas sobre essa instalação).

Como estou executando no Ubuntu (o release Intrepid), eu uso o utilitário apt
para usar a distribuição Hadoop. Esse processo é muito simples e me
permite usar o pacote binário sem os detalhes adicionais de fazer o
download e criar a origem. Primeiro, informo apt sobre o site Cloudera. Em seguida, crio um novo arquivo em /etc/apt/sources.list.d/cloudera.list e adiciono o seguinte texto:

deb http://archive.cloudera.com/debian intrepid-cdh3 contrib<br />deb-src http://archive.cloudera.com/debian intrepid-cdh3 contrib

Se estiver executando Jaunty ou outro release, substitua intrepid pelo nome do release específico (o suporte atual inclui Hardy, Intrepid, Jaunty, Karmic e Lenny).

Em seguida, eu uso o apt-key de Cloudera para validar o pacote transferido por download:

$ curl -s http://archive.cloudera.com/debian/archive.key | \<br />sudo apt-key add - sudo apt-get update

E instalo Hadoop para uma configuração pseudodistribuída (todos os daemons de Hadoop são executados em um único host):

$ sudo apt-get install hadoop-0.20-conf-pseudo<br />$

Observe que essa configuração está em torno de 23 MB (não incluindo nenhum outro pacote que apt
obtenha que possa não estar presente). Essa instalação é ideal para ser
feita com Hadoop e conhecer mais de seus elementos e interfaces.

Por fim, eu configuro SSH sem passphrase. Se tentar usar ssh localhost,
e uma passphrase for solicitada, será preciso seguir as próximas
etapas. Eu considero que esta é uma caixa Hadoop dedicada, visto que
esta etapa tem algumas implicações de segurança (consulte a Listagem 1).

Listagem 1. Configuração de SSH sem passphrase

$ sudo su -<br /># ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa<br /># cat ~/.ssh/id_dsa.pub >> ~/.ssh/authorized_keys

Uma observação final é que é preciso verificar se o host tem espaço de
armazenamento disponível para o datanode (o cache). O armazenamento
insuficiente se manifesta de maneiras estranhas (como erros que indicam a
incapacidade de replicação para um nó).

Iniciando o Hadoop

Agora,
você está pronto para iniciar o Hadoop, o que pode ser feito
efetivamente com a inicialização de cada daemon do Hadoop. Mas,
primeiro, formate o sistema de arquivos Hadoop (HDFS) usando o comando hadoop. O comando hadoop tem inúmeros usos, exploraremos alguns deles em breve.

Primeiro,
solicite o namenode para formatar o sistema de arquivos DFS. Isso é
feito como parte da instalação, mas é útil saber caso alguma vez precise
gerar um sistema de arquivos limpo.

# hadoop-0.20 namenode -format

Após a aceitação da solicitação, o sistema de arquivos será formatado, e
algumas informações, retornadas. Em seguida, inicie os daemons do
Hadoop. O Hadoop inicia cinco daemons nesta configuração
pseudodistribuída: namenode, secondarynamenode, datanode, jobtracker e
tasktracker.

Quando cada um dos daemons tiver sido iniciado, você verá
uma pequena quantidade de texto emitida por todos (identificando onde
seus logs estão armazenados). Cada daemon está sendo iniciado para
execução em segundo plano (como um daemon). A Figura 1 mostra como o nó
pseudodistribuído ficará quando a inicialização estiver concluída.

Figura 1. Configuração de Hadoop pseudodistribuída

Hadoop fornece algumas ferramentas auxiliares para simplificar a sua
inicialização. Essas ferramentas estão categorizadas como início (como start-dfs) e parada (como stop-dfs). O seguinte script curto mostra como iniciar o nó Hadoop:

# /usr/lib/hadoop-0.20/bin/start-dfs.sh<br /># /usr/lib/hadoop-0.20/bin/start-mapred.sh<br />#

Para verificar se os daemons estão em execução, você poderá usar o comando jps (que é um utilitário ps para os processos de JVM). Esse comando lista os cinco daemons e seus identificadores de processo.

Agora que os daemons do Hadoop estão em execução, voltemos a eles para apresentar o que cada um faz na estrutura do Hadoop. O namenode
é o servidor mestre no Hadoop, gerencia o namespace do sistema de
arquivos e acessa os arquivos armazenados no cluster. Há também um namenode secundário,
que não é um daemon redundante para o namenode, mas, em vez disso,
oferece tarefas de ponto de verificação e manutenção do período. Você
encontrará um namenode e um namenode secundário em um cluster do Hadoop.

O datanode gerencia o armazenamento anexado a um
nó, no qual pode haver vários nós em um cluster. Cada nó que armazena
dados terá um daemon datanode em execução.

Por fim, cada cluster terá um único jobtracker responsável por programar o trabalho nos datanodes e um tasktracker
por datanode que realiza o trabalho real. O jobtracker e o tasktracker
se comportam em uma disposição principal-escravo, em que o jobtracker
distribui trabalho nos datanodes, e o tasktracker executa a tarefa. O
jobtracker também valida o trabalho solicitado e, se um datanode falhar
por algum motivo, programa novamente a tarefa anterior.

Nessa configuração simples, todos os nós simplesmente residem no mesmo nó (consulte a Figura 1).
Mas, a partir da discussão anterior, é fácil ver como o Hadoop oferece
processamento de trabalho paralelo. Embora a arquitetura seja simples, o
Hadoop é uma forma fácil de distribuir dados, balancear carga e
processar paralelamente grandes quantidades de dados de uma forma
tolerante a falhas.

Inspecionando HDFS

É possível realizar vários testes para garantir que o Hadoop funcione
normalmente (pelo menos o namenode). Sabendo que todos os seus
processos estão disponíveis, você poderá usar o comando hadoop para inspecionar o namespace local (consulte a Listagem 2).

Listagem 2. Verificando o acesso ao HDFS

# <strong>hadoop-0.20 fs -ls /</strong><br />Found 2 items<br />drwxr-xr-x   - root supergroup          0 2010-04-29 16:38 /user<br />drwxr-xr-x   - root supergroup          0 2010-04-29 16:28 /var<br />#

A partir disso, será possível ver que o namenode está ativo e
preparado para operar o namespace local. Observe que está sendo usado um
comando denominado hadoop-0.20 para inspecionar o sistema
de arquivos. Esse utilitário é o modo como você interage com o cluster
do Hadoop, desde a inspeção do sistema de arquivos até a execução de
tarefas no cluster.

Observe a estrutura do comando aqui: após
especificar o utilitário hadoop-0.20, defina um comando
(nesse caso, o shell do sistema de arquivos genérico) e uma ou mais
opções (nesse caso, solicite uma lista de arquivos usando ls). Como o hadoop-0.20
é uma das interfaces principais para o cluster do Hadoop, você usará
esse utilitário um pouco neste artigo. A Listagem 3 fornece algumas
operações de sistema de arquivos adicionais com as quais é possível
explorar essa interface um pouco mais (criar um novo subdiretório
denominado test que lista seu conteúdo e depois o remove).

Listagem 3. Explorando a manipulação do sistema de arquivos no Hadoop

# <strong>hadoop-0.20 fs -mkdir test</strong><br /># <strong>hadoop-0.20 fs -ls test</strong><br /># <strong>hadoop-0.20 fs -rmr test</strong><br />Deleted hdfs://localhost/user/root/test<br /># 

Testando o Hadoop

Agora
que instalamos o Hadoop e testamos a interface básica do seu sistema de
arquivos, chegou a hora de testar o Hadoop em um aplicativo real. Neste
exemplo, veja o processo MapReduce em um pequeno conjunto de dados.

Map e reduce têm nomes de funções da programação funcional, mas fornecem a capacidade principal para redução de dados. Map
se refere ao processo de dividir a entrada em um conjunto menor de
subproblemas para processamento (em que esses subproblemas são
distribuídos a trabalhadores paralelos). Reduce se refere à montagem de respostas a partir de subproblemas em um único conjunto de saída. Observe que não defini o que processamento
significa, visto que a estrutura permite que você mesmo defina isso.
Canonical MapReduce é o cálculo da frequência de palavras em um conjunto
de documentos.

De acordo com a discussão anterior, haverá um
conjunto de entradas e um conjunto resultante de saídas. A primeira
etapa é criar um subdiretório de entrada no sistema de arquivos em que o
trabalho será colocado. Faça isso usando:

# hadoop-0.20 fs -mkdir input

Em seguida, coloque algum trabalho no subdiretório de entrada. Nesse caso, use o comando put,
que move um arquivo do sistema de arquivos local para o HDFS (consulte a
Listagem 4). Observe o formato abaixo, que move o arquivo de origem
para o subdiretório HDFS (entrada). Quando terminar, você terá dois
arquivos de texto no HDFS prontos para serem processados.

Listagem 4. Movendo arquivos para o HDFS

# hadoop-0.20 fs -put /usr/src/linux-source-2.6.27/Doc*/memory-barriers.txt  input<br /># hadoop-0.20 fs -put /usr/src/linux-source-2.6.27/Doc*/rt-mutex-design.txt  input<br />#

Em seguida, é possível verificar a presença dos arquivos usando o comando ls (consulte a Listagem 5).

Listagem 5. Verificando arquivos no HDFS

# <strong>hadoop-0.20 fs -ls input</strong><br />Found 2 items<br />-rw-r--r--  1 root supergroup 78031 2010-04-29 17:35 /user/root/input/memory-barriers.txt<br />-rw-r--r--  1 root supergroup 33567 2010-04-29 17:36 /user/root/input/rt-mutex-design.txt <br />#

Com o seu trabalho armazenado com segurança no HDFS, execute a função
MapReduce. Essa função requer um único comando, mas uma solicitação
longa, como mostrado na Listagem 6. Esse comando solicita a execução de
um JAR. Ele realmente implementa vários recursos, mas este exemplo
enfoca wordcount.

O daemon jobtracker solicita que o
datanode execute a tarefa MapReduce, que resulta em uma quantidade
considerável de saída (menor aqui, porque estão sendo processados apenas
dois arquivos). Ele mostra o progresso do mapa e as funções de redução e
fornece algumas estatísticas úteis com relação à E/S para o sistema de
arquivos e o processamento de registros.

Listagem 6. Executando uma tarefa MapReduce para frequência de palavras (contagem de palavras)

# <strong>hadoop-0.20 jar /usr/lib/hadoop-0.20/hadoop-0.20.2+228-examples.jar \<br />wordcount input output</strong><br />10/04/29 17:36:49 INFO input.FileInputFormat: Total input paths to process : 2<br />10/04/29 17:36:49 INFO mapred.JobClient: Running job: job_201004291628_0009<br />10/04/29 17:36:50 INFO mapred.JobClient:  map 0% reduce 0%<br />10/04/29 17:37:00 INFO mapred.JobClient:  map 100% reduce 0%<br />10/04/29 17:37:06 INFO mapred.JobClient:  map 100% reduce 100%<br />10/04/29 17:37:08 INFO mapred.JobClient: Job complete: job_201004291628_0009<br />10/04/29 17:37:08 INFO mapred.JobClient: Counters: 17<br />10/04/29 17:37:08 INFO mapred.JobClient:   Job Counters <br />10/04/29 17:37:08 INFO mapred.JobClient:     Launched reduce tasks=1<br />10/04/29 17:37:08 INFO mapred.JobClient:     Launched map tasks=2<br />10/04/29 17:37:08 INFO mapred.JobClient:     Data-local map tasks=2<br />10/04/29 17:37:08 INFO mapred.JobClient:   FileSystemCounters<br />10/04/29 17:37:08 INFO mapred.JobClient:     FILE_BYTES_READ=47556<br />10/04/29 17:37:08 INFO mapred.JobClient:     HDFS_BYTES_READ=111598<br />10/04/29 17:37:08 INFO mapred.JobClient:     FILE_BYTES_WRITTEN=95182<br />10/04/29 17:37:08 INFO mapred.JobClient:     HDFS_BYTES_WRITTEN=30949<br />10/04/29 17:37:08 INFO mapred.JobClient:   Map-Reduce Framework<br />10/04/29 17:37:08 INFO mapred.JobClient:     Reduce input groups=2974<br />10/04/29 17:37:08 INFO mapred.JobClient:     Combine output records=3381<br />10/04/29 17:37:08 INFO mapred.JobClient:     Map input records=2937<br />10/04/29 17:37:08 INFO mapred.JobClient:     Reduce shuffle bytes=47562<br />10/04/29 17:37:08 INFO mapred.JobClient:     Reduce output records=2974<br />10/04/29 17:37:08 INFO mapred.JobClient:     Spilled Records=6762<br />10/04/29 17:37:08 INFO mapred.JobClient:     Map output bytes=168718<br />10/04/29 17:37:08 INFO mapred.JobClient:     Combine input records=17457<br />10/04/29 17:37:08 INFO mapred.JobClient:     Map output records=17457<br />10/04/29 17:37:08 INFO mapred.JobClient:     Reduce input records=3381 

Com o processamento completo, inspecione o resultado. Lembre-se de que a finalidade da tarefa é calcular quantas vezes as palavras ocorreram nos arquivos de entrada. Essa saída é emitida como um arquivo de tuplas, representando a palavra e quantas vezes ela aparece na entrada. É possível usar o comando cat (após a localização do arquivo de saída particular) por meio do utilitário hadoop-0.20 para emitir esses dados (consulte a Listagem 7).

Listagem 7. Revisando a saída da operação de contagem de palavras de MapReduce

# <strong>hadoop-0.20 fs -ls /user/root/output</strong><br />Found 2 items<br />drwxr-xr-x   - root supergroup          0 2010-04-29 17:36 /user/root/output/_logs<br />-rw-r--r--   1 root supergroup      30949 2010-04-29 17:37 /user/root/output/part-r-00000<br />#  <br /># <strong>hadoop-0.20 fs -cat output/part-r-00000 | head -13</strong><br />!= 1<br />"Atomic 2<br />"Cache 2<br />"Control 1<br />"Examples 1<br />"Has 7<br />"Inter-CPU 1<br />"LOAD 1<br />"LOCK" 1<br />"Locking 1<br />"Locks 1<br />"MMIO 1<br />"Pending 5<br />#

É possível também extrair o arquivo do HDFS usando o utilitário hadoop-0.20 (consulte a Listagem 8). Isso pode ser feito facilmente com o utilitário get (análogo ao put executado antes para escrever arquivos no HDFS). Para a operação get, especifique o arquivo no HDFS a ser extraído (a partir do subdiretório de saída) e o nome do arquivo a ser gravado no sistema de arquivos local (output.txt).

Listagem 8. Extraindo a saída de HDFS

# <strong>hadoop-0.20 fs -get output/part-r-00000 output.txt</strong><br /># <strong>cat output.txt | head -5</strong><br />!= 1<br />"Atomic 2<br />"Cache 2<br />"Control 1<br />"Examples 1<br /># 

Vamos observar outro exemplo usando o mesmo JAR, mas com um uso diferente (aqui, será explorado um grep paralelo). Para esse teste, use os seus arquivos de entrada existentes, mas remova o subdiretório de saída para recriá-lo para o teste:

# <strong>hadoop-0.20 fs -rmr output</strong><br />Deleted hdfs://localhost/user/root/output

Em seguida, solicite a tarefa MapReduce para grep. Nesse caso, o grep é executado em paralelo (o map), e os resultados de grep são combinados (o reduce). A Listagem 9 mostra a saída desse modelo de uso (mas, nesse caso, uma parte da saída foi suprimida para fins de brevidade). Observe a solicitação de comando aqui, em que sua solicitação é um grep que obtém a entrada do subdiretório chamado input e coloca o resultado em um subdiretório chamado output. O parâmetro final é a cadeia de caracteres sendo pesquisada (nesse caso, ‘kernel’).

Listagem 9. Executando uma tarefa MapReduce para contagem de pesquisa de palavras (grep)

# <strong>hadoop-0.20 jar /usr/lib/hadoop/hadoop-0.20.2+228-examples.jar \<br />grep input output 'kernel'</strong><br />10/04/30 09:22:29 INFO mapred.FileInputFormat: Total input paths to process : 2<br />10/04/30 09:22:30 INFO mapred.JobClient: Running job: job_201004291628_0010<br />10/04/30 09:22:31 INFO mapred.JobClient:  map 0% reduce 0%<br />10/04/30 09:22:42 INFO mapred.JobClient:  map 66% reduce 0%<br />10/04/30 09:22:45 INFO mapred.JobClient:  map 100% reduce 0%<br />10/04/30 09:22:54 INFO mapred.JobClient:  map 100% reduce 100%<br />10/04/30 09:22:56 INFO mapred.JobClient: Job complete: job_201004291628_0010<br />10/04/30 09:22:56 INFO mapred.JobClient: Counters: 18<br />10/04/30 09:22:56 INFO mapred.JobClient:   Job Counters <br />10/04/30 09:22:56 INFO mapred.JobClient:     Launched reduce tasks=1<br />10/04/30 09:22:56 INFO mapred.JobClient:     Launched map tasks=3<br />10/04/30 09:22:56 INFO mapred.JobClient:     Data-local map tasks=3<br />10/04/30 09:22:56 INFO mapred.JobClient:   FileSystemCounters<br />10/04/30 09:22:56 INFO mapred.JobClient:     FILE_BYTES_READ=57<br />10/04/30 09:22:56 INFO mapred.JobClient:     HDFS_BYTES_READ=113144<br />10/04/30 09:22:56 INFO mapred.JobClient:     FILE_BYTES_WRITTEN=222<br />10/04/30 09:22:56 INFO mapred.JobClient:     HDFS_BYTES_WRITTEN=109<br />...<br />10/04/30 09:23:14 INFO mapred.JobClient:     Map output bytes=15<br />10/04/30 09:23:14 INFO mapred.JobClient:     Map input bytes=23<br />10/04/30 09:23:14 INFO mapred.JobClient:     Combine input records=0<br />10/04/30 09:23:14 INFO mapred.JobClient:     Map output records=1<br />10/04/30 09:23:14 INFO mapred.JobClient:     Reduce input records=1<br />#

Com a tarefa completa, inspecione o diretório de saída (para identificar o arquivo de resultados) e execute uma operação de sistema de arquivos cat para exibir seu conteúdo (consulte a Listagem 10).

Listagem 10. Inspecionando a saída da tarefa MapReduce

# <strong>hadoop-0.20 fs -ls output</strong><br />Found 2 items<br />drwxr-xr-x  - root supergroup    0 2010-04-30 09:22 /user/root/output/_logs<br />-rw-r--r--  1 root supergroup   10 2010-04-30 09:23 /user/root/output/part-00000<br /># <strong>hadoop-0.20 fs -cat output/part-00000</strong><br />17 kernel<br /># 

Interfaces baseadas na Web

Vimos como inspecionar o HDFS, mas, se estiver procurando informações sobre a operação do Hadoop, você achará as interfaces da Web úteis. Lembre-se de que, no topo do cluster do Hadoop, está o namenode, que gerencia o HDFS.

É possível explorar detalhes de alto nível do sistema de arquivos (como espaço disponível e usado e datanodes disponíveis), bem como executar trabalhos por meio de http://localhost:50070. Aprofunde-se no jobtracker (status da tarefa) por meio de http://localhost:50030. Observe que, em ambos os casos, faz-se referência a localhost, porque todos os daemons estão em execução no mesmo host.

Indo mais longe

Este artigo mostrou a instalação e a configuração inicial de um cluster Hadoop (pseudodistribuído) simples (nesse caso, usando a distribuição de Cloudera para Hadoop). Escolhi essa distribuição particular porque ela simplificou a instalação e a configuração inicial de Hadoop. É possível encontrar várias distribuições para Hadoop (incluindo a origem) em apache.org.

Mas e se você não tiver os recursos de hardware suficientes para dimensionar o cluster de Hadoop para as suas necessidades específicas? Constatamos que o Hadoop é muito conhecido, portanto, é possível executá-lo facilmente nas infraestruturas de computação em nuvem usando VMs de Hadoop criadas previamente e servidores de lease. Amazon oferece Amazon Machine Images (AMIs), bem como recursos de computação no Amazon Elastic Compute Cloud (Amazon EC2). Adicionalmente, a Microsoft anunciou recentemente o suporte para Hadoop em sua Windows Azure Services Platform.

Neste artigo, é fácil ver como Hadoop torna a computação distribuída simples para o processamento de grandes conjuntos de dados. O próximo artigo desta série mostrará como configurar o Hadoop em um cluster com vários nós com exemplos adicionais. Até a próxima!

Recursos

Aprender

Obter produtos e tecnologias

*

artigo publicado originalmente no developerWorks Brasil, por M. Tim Jones


M. Tim Jones é arquiteto de firmware integrado e autor das obras Artificial Intelligence: A Systems Approach, GNU/Linux Application Programming (agora, na segunda edição), AI Application Programming (na segunda edição) e BSD Sockets Programming from a Multilanguage Perspective.
Sua experiência em engenharia vai desde o desenvolvimento de kernels
para espaçonaves geossíncronas até a arquitetura de sistemas integrados e
o desenvolvimento de protocolos de rede. Tim é engenheiro consultor da
Emulex Corp. em Longmont, Colorado.

é o portal de tecnologia da IBM para profissionais de TI de todo o mundo. Colabore, aprenda a criar aplicativos inovadores e compreenda tecnologias avançadas. Acesse mais artigos e tutoriais em www.ibm.com/developerworks/br

Ver perfil