MapReduce - Guia rápido

MapReduce é um modelo de programação para escrever aplicativos que podem processar Big Data em paralelo em vários nós. MapReduce fornece recursos analíticos para analisar grandes volumes de dados complexos.

O que é Big Data?

Big Data é uma coleção de grandes conjuntos de dados que não podem ser processados ​​usando técnicas tradicionais de computação. Por exemplo, o volume de dados que o Facebook ou Youtube precisam para coletar e gerenciar diariamente pode cair na categoria de Big Data. No entanto, Big Data não envolve apenas escala e volume, ele também envolve um ou mais dos seguintes aspectos - Velocidade, Variedade, Volume e Complexidade.

Por que MapReduce?

Os sistemas corporativos tradicionais normalmente possuem um servidor centralizado para armazenar e processar dados. A ilustração a seguir descreve uma visão esquemática de um sistema empresarial tradicional. O modelo tradicional certamente não é adequado para processar grandes volumes de dados escalonáveis ​​e não pode ser acomodado por servidores de banco de dados padrão. Além disso, o sistema centralizado cria um grande gargalo ao processar vários arquivos simultaneamente.

O Google resolveu esse problema de gargalo usando um algoritmo chamado MapReduce. O MapReduce divide uma tarefa em pequenas partes e as atribui a vários computadores. Posteriormente, os resultados são coletados em um local e integrados para formar o conjunto de dados de resultados.

Como funciona o MapReduce?

O algoritmo MapReduce contém duas tarefas importantes, ou seja, Mapear e Reduzir.

  • A tarefa Map pega um conjunto de dados e os converte em outro conjunto de dados, onde os elementos individuais são divididos em tuplas (pares chave-valor).

  • A tarefa Reduce pega a saída do Map como uma entrada e combina essas tuplas de dados (pares de valores-chave) em um conjunto menor de tuplas.

A tarefa de redução é sempre executada após o trabalho de mapa.

Vamos agora dar uma olhada em cada uma das fases e tentar entender seu significado.

  • Input Phase - Aqui temos um leitor de registros que traduz cada registro em um arquivo de entrada e envia os dados analisados ​​para o mapeador na forma de pares chave-valor.

  • Map - Map é uma função definida pelo usuário, que pega uma série de pares de valores-chave e processa cada um deles para gerar zero ou mais pares de valores-chave.

  • Intermediate Keys - Os pares de valores-chave gerados pelo mapeador são conhecidos como chaves intermediárias.

  • Combiner- Um combinador é um tipo de Redutor local que agrupa dados semelhantes da fase do mapa em conjuntos identificáveis. Ele pega as chaves intermediárias do mapeador como entrada e aplica um código definido pelo usuário para agregar os valores em um pequeno escopo de um mapeador. Não faz parte do algoritmo MapReduce principal; é opcional.

  • Shuffle and Sort- A tarefa do Redutor começa com a etapa de ordem aleatória e classificação. Ele baixa os pares de valores-chave agrupados na máquina local, onde o Redutor está sendo executado. Os pares de valores-chave individuais são classificados por chave em uma lista de dados maior. A lista de dados agrupa as chaves equivalentes para que seus valores possam ser iterados facilmente na tarefa do Redutor.

  • Reducer- O Redutor recebe os dados emparelhados de valores-chave agrupados como entrada e executa uma função de Redutor em cada um deles. Aqui, os dados podem ser agregados, filtrados e combinados de várias maneiras e requer uma ampla variedade de processamento. Quando a execução termina, ele fornece zero ou mais pares de valores-chave para a etapa final.

  • Output Phase - Na fase de saída, temos um formatador de saída que traduz os pares de valores-chave finais da função Redutor e os grava em um arquivo usando um gravador.

Vamos tentar entender as duas tarefas Map & f Reduce com a ajuda de um pequeno diagrama -

MapReduce-Example

Vamos dar um exemplo do mundo real para compreender o poder do MapReduce. O Twitter recebe cerca de 500 milhões de tuítes por dia, o que é quase 3.000 tuítes por segundo. A ilustração a seguir mostra como o Tweeter gerencia seus tweets com a ajuda do MapReduce.

Conforme mostrado na ilustração, o algoritmo MapReduce executa as seguintes ações -

  • Tokenize - Tokeniza os tweets em mapas de tokens e os grava como pares de valores-chave.

  • Filter - Filtra palavras indesejadas dos mapas de tokens e grava os mapas filtrados como pares de valores-chave.

  • Count - Gera um contador de tokens por palavra.

  • Aggregate Counters - Prepara um agregado de valores de contador semelhantes em pequenas unidades gerenciáveis.

O algoritmo MapReduce contém duas tarefas importantes, ou seja, Mapear e Reduzir.

  • A tarefa do mapa é feita por meio da Classe Mapper
  • A tarefa de redução é feita por meio da Classe Redutor.

A classe Mapper pega a entrada, cria tokens, mapeia e classifica. A saída da classe Mapper é usada como entrada pela classe Reducer, que por sua vez pesquisa pares correspondentes e os reduz.

MapReduce implementa vários algoritmos matemáticos para dividir uma tarefa em pequenas partes e atribuí-las a vários sistemas. Em termos técnicos, o algoritmo MapReduce ajuda a enviar as tarefas Map & Reduce para os servidores apropriados em um cluster.

Esses algoritmos matemáticos podem incluir o seguinte -

  • Sorting
  • Searching
  • Indexing
  • TF-IDF

Ordenação

A classificação é um dos algoritmos MapReduce básicos para processar e analisar dados. MapReduce implementa algoritmo de classificação para classificar automaticamente os pares de valor-chave de saída do mapeador por suas chaves.

  • Os métodos de classificação são implementados na própria classe do mapeador.

  • Na fase de ordem aleatória e classificação, após tokenizar os valores na classe do mapeador, o Context classe (classe definida pelo usuário) coleta as chaves de valor correspondentes como uma coleção.

  • Para coletar pares de valores-chave semelhantes (chaves intermediárias), a classe Mapper tem a ajuda de RawComparator classe para classificar os pares de valor-chave.

  • O conjunto de pares de valores-chave intermediários para um determinado Redutor é classificado automaticamente pelo Hadoop para formar valores-chave (K2, {V2, V2, ...}) antes de serem apresentados ao Redutor.

Procurando

A pesquisa desempenha um papel importante no algoritmo MapReduce. Auxilia na fase combinadora (opcional) e na fase redutora. Vamos tentar entender como Searching funciona com a ajuda de um exemplo.

Exemplo

O exemplo a seguir mostra como MapReduce emprega o algoritmo de pesquisa para descobrir os detalhes do funcionário que recebe o maior salário em um determinado conjunto de dados de funcionários.

  • Vamos supor que temos dados de funcionários em quatro arquivos diferentes - A, B, C e D. Suponhamos também que haja registros de funcionários duplicados em todos os quatro arquivos devido à importação de dados de funcionários de todas as tabelas do banco de dados repetidamente. Veja a ilustração a seguir.

  • The Map phaseprocessa cada arquivo de entrada e fornece os dados do funcionário em pares de valores-chave (<k, v>: <nome emp, salário>). Veja a ilustração a seguir.

  • The combiner phase(técnica de pesquisa) aceitará a entrada da fase de Mapa como um par de valores-chave com o nome do funcionário e o salário. Usando a técnica de pesquisa, o combinador irá verificar todo o salário do funcionário para encontrar o funcionário com maior salário em cada arquivo. Veja o seguinte trecho.

<k: employee name, v: salary>
Max= the salary of an first employee. Treated as max salary

if(v(second employee).salary > Max){
   Max = v(salary);
}

else{
   Continue checking;
}

O resultado esperado é o seguinte -

<satish, 26.000>

<gopal, 50000>

<kiran, 45000>

<manisha, 45000>

  • Reducer phase- De cada arquivo, você encontrará o funcionário com maior salário. Para evitar redundância, verifique todos os pares <k, v> e elimine entradas duplicadas, se houver. O mesmo algoritmo é usado entre os quatro pares <k, v>, que vêm de quatro arquivos de entrada. O resultado final deve ser o seguinte -

<gopal, 50000>

Indexando

Normalmente, a indexação é usada para apontar para um dado específico e seu endereço. Ele executa a indexação de lote nos arquivos de entrada para um mapeador específico.

A técnica de indexação normalmente usada no MapReduce é conhecida como inverted index.Motores de busca como Google e Bing usam técnicas de indexação invertida. Vamos tentar entender como a indexação funciona com a ajuda de um exemplo simples.

Exemplo

O texto a seguir é a entrada para indexação invertida. Aqui, T [0], T [1] e t [2] são os nomes dos arquivos e seu conteúdo está entre aspas duplas.

T[0] = "it is what it is"
T[1] = "what is it"
T[2] = "it is a banana"

Depois de aplicar o algoritmo de indexação, obtemos a seguinte saída -

"a": {2}
"banana": {2}
"is": {0, 1, 2}
"it": {0, 1, 2}
"what": {0, 1}

Aqui, "a": {2} implica que o termo "a" aparece no arquivo T [2]. Da mesma forma, "é": {0, 1, 2} implica que o termo "é" aparece nos arquivos T [0], T [1] e T [2].

TF-IDF

TF-IDF é um algoritmo de processamento de texto abreviação de Term Frequency - Inverse Document Frequency. É um dos algoritmos comuns de análise da web. Aqui, o termo 'frequência' se refere ao número de vezes que um termo aparece em um documento.

Frequência de termo (TF)

Ele mede a frequência com que um determinado termo ocorre em um documento. É calculado pelo número de vezes que uma palavra aparece em um documento dividido pelo número total de palavras nesse documento.

TF(the) = (Number of times term the ‘the’ appears in a document) / (Total number of terms in the document)

Frequência inversa do documento (IDF)

Ele mede a importância de um termo. É calculado pelo número de documentos na base de dados de texto dividido pelo número de documentos onde um termo específico aparece.

Ao calcular TF, todos os termos são considerados igualmente importantes. Isso significa que TF conta a frequência do termo para palavras normais como "é", "a", "o que", etc. Assim, precisamos conhecer os termos frequentes enquanto aumentamos os raros, calculando o seguinte -

IDF(the) = log_e(Total number of documents / Number of documents with term ‘the’ in it).

O algoritmo é explicado a seguir com a ajuda de um pequeno exemplo.

Exemplo

Considere um documento contendo 1000 palavras, em que a palavra hiveaparece 50 vezes. O TF parahive é então (50/1000) = 0,05.

Agora, suponha que temos 10 milhões de documentos e a palavra hiveaparece em 1000 deles. Então, o IDF é calculado como log (10.000.000 / 1.000) = 4.

O peso TF-IDF é o produto dessas quantidades - 0,05 × 4 = 0,20.

MapReduce funciona apenas em sistemas operacionais com estilo Linux e vem embutido com uma estrutura Hadoop. Precisamos realizar as seguintes etapas para instalar a estrutura do Hadoop.

Verificando a instalação JAVA

Java deve ser instalado em seu sistema antes de instalar o Hadoop. Use o seguinte comando para verificar se você possui o Java instalado em seu sistema.

$ java –version

Se o Java já estiver instalado em seu sistema, você verá a seguinte resposta -

java version "1.7.0_71"
Java(TM) SE Runtime Environment (build 1.7.0_71-b13)
Java HotSpot(TM) Client VM (build 25.0-b02, mixed mode)

Caso você não tenha o Java instalado em seu sistema, siga as etapas abaixo.

Instalando Java

Passo 1

Baixe a versão mais recente do Java no seguinte link - este link .

Após o download, você pode localizar o arquivo jdk-7u71-linux-x64.tar.gz na pasta Downloads.

Passo 2

Use os comandos a seguir para extrair o conteúdo de jdk-7u71-linux-x64.gz.

$ cd Downloads/
$ ls jdk-7u71-linux-x64.gz $ tar zxf jdk-7u71-linux-x64.gz
$ ls
jdk1.7.0_71 jdk-7u71-linux-x64.gz

etapa 3

Para disponibilizar o Java a todos os usuários, você deve movê-lo para o local “/ usr / local /”. Vá para o root e digite os seguintes comandos -

$ su
password:
# mv jdk1.7.0_71 /usr/local/java
# exit

Passo 4

Para configurar as variáveis ​​PATH e JAVA_HOME, adicione os seguintes comandos ao arquivo ~ / .bashrc.

export JAVA_HOME=/usr/local/java
export PATH=$PATH:$JAVA_HOME/bin

Aplique todas as alterações ao sistema em execução atual.

$ source ~/.bashrc

Etapa 5

Use os seguintes comandos para configurar alternativas Java -

# alternatives --install /usr/bin/java java usr/local/java/bin/java 2

# alternatives --install /usr/bin/javac javac usr/local/java/bin/javac 2

# alternatives --install /usr/bin/jar jar usr/local/java/bin/jar 2

# alternatives --set java usr/local/java/bin/java

# alternatives --set javac usr/local/java/bin/javac

# alternatives --set jar usr/local/java/bin/jar

Agora verifique a instalação usando o comando java -version do terminal.

Verificando a instalação do Hadoop

O Hadoop deve ser instalado em seu sistema antes de instalar o MapReduce. Vamos verificar a instalação do Hadoop usando o seguinte comando -

$ hadoop version

Se o Hadoop já estiver instalado em seu sistema, você receberá a seguinte resposta -

Hadoop 2.4.1
--
Subversion https://svn.apache.org/repos/asf/hadoop/common -r 1529768
Compiled by hortonmu on 2013-10-07T06:28Z
Compiled with protoc 2.5.0
From source with checksum 79e53ce7994d1628b240f09af91e1af4

Se o Hadoop não estiver instalado em seu sistema, prossiga com as etapas a seguir.

Baixando Hadoop

Baixe o Hadoop 2.4.1 da Apache Software Foundation e extraia seu conteúdo usando os comandos a seguir.

$ su
password:
# cd /usr/local
# wget http://apache.claz.org/hadoop/common/hadoop-2.4.1/
hadoop-2.4.1.tar.gz
# tar xzf hadoop-2.4.1.tar.gz
# mv hadoop-2.4.1/* to hadoop/
# exit

Instalando o Hadoop no modo Pseudo Distribuído

As etapas a seguir são usadas para instalar o Hadoop 2.4.1 no modo pseudo distribuído.

Etapa 1 - Configurando o Hadoop

Você pode definir variáveis ​​de ambiente Hadoop anexando os seguintes comandos ao arquivo ~ / .bashrc.

export HADOOP_HOME=/usr/local/hadoop
export HADOOP_MAPRED_HOME=$HADOOP_HOME
export HADOOP_COMMON_HOME=$HADOOP_HOME export HADOOP_HDFS_HOME=$HADOOP_HOME
export YARN_HOME=$HADOOP_HOME export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native
export PATH=$PATH:$HADOOP_HOME/sbin:$HADOOP_HOME/bin

Aplique todas as alterações ao sistema em execução atual.

$ source ~/.bashrc

Etapa 2 - Configuração do Hadoop

Você pode encontrar todos os arquivos de configuração do Hadoop no local “$ HADOOP_HOME / etc / hadoop”. Você precisa fazer as alterações adequadas nesses arquivos de configuração de acordo com sua infraestrutura Hadoop.

$ cd $HADOOP_HOME/etc/hadoop

Para desenvolver programas Hadoop usando Java, você deve redefinir as variáveis ​​de ambiente Java em hadoop-env.sh substituindo o valor JAVA_HOME pelo local do Java em seu sistema.

export JAVA_HOME=/usr/local/java

Você deve editar os seguintes arquivos para configurar o Hadoop -

  • core-site.xml
  • hdfs-site.xml
  • yarn-site.xml
  • mapred-site.xml

core-site.xml

core-site.xml contém as seguintes informações−

  • Número da porta usado para instância Hadoop
  • Memória alocada para o sistema de arquivos
  • Limite de memória para armazenar os dados
  • Tamanho dos buffers de leitura / gravação

Abra o core-site.xml e adicione as seguintes propriedades entre as tags <configuration> e </configuration>.

<configuration>
   <property>
      <name>fs.default.name</name>
      <value>hdfs://localhost:9000 </value>
   </property>
</configuration>

hdfs-site.xml

hdfs-site.xml contém as seguintes informações -

  • Valor dos dados de replicação
  • O caminho do namenode
  • O caminho do datanode de seus sistemas de arquivos locais (o lugar onde você deseja armazenar o infra Hadoop)

Vamos supor os seguintes dados.

dfs.replication (data replication value) = 1

(In the following path /hadoop/ is the user name.
hadoopinfra/hdfs/namenode is the directory created by hdfs file system.)
namenode path = //home/hadoop/hadoopinfra/hdfs/namenode

(hadoopinfra/hdfs/datanode is the directory created by hdfs file system.)
datanode path = //home/hadoop/hadoopinfra/hdfs/datanode

Abra este arquivo e adicione as seguintes propriedades entre as marcas <configuration>, </configuration>.

<configuration>

   <property>
      <name>dfs.replication</name>
      <value>1</value>
   </property>
   
   <property>
      <name>dfs.name.dir</name>
      <value>file:///home/hadoop/hadoopinfra/hdfs/namenode</value>
   </property>
   
   <property>
      <name>dfs.data.dir</name>
      <value>file:///home/hadoop/hadoopinfra/hdfs/datanode </value>
   </property>
   
</configuration>

Note - No arquivo acima, todos os valores de propriedade são definidos pelo usuário e você pode fazer alterações de acordo com sua infraestrutura Hadoop.

yarn-site.xml

Este arquivo é usado para configurar o yarn no Hadoop. Abra o arquivo yarn-site.xml e adicione as seguintes propriedades entre as tags <configuration>, </configuration>.

<configuration>
   <property>
      <name>yarn.nodemanager.aux-services</name>
      <value>mapreduce_shuffle</value>
   </property>
</configuration>

mapred-site.xml

Este arquivo é usado para especificar a estrutura MapReduce que estamos usando. Por padrão, o Hadoop contém um modelo de yarn-site.xml. Primeiro de tudo, você precisa copiar o arquivo mapred-site.xml.template para o arquivo mapred-site.xml usando o seguinte comando.

$ cp mapred-site.xml.template mapred-site.xml

Abra o arquivo mapred-site.xml e adicione as seguintes propriedades entre as marcas <configuration>, </configuration>.

<configuration>
   <property>
      <name>mapreduce.framework.name</name>
      <value>yarn</value>
   </property>
</configuration>

Verificando a instalação do Hadoop

As etapas a seguir são usadas para verificar a instalação do Hadoop.

Etapa 1 - Configuração do Nó de Nome

Configure o namenode usando o comando “hdfs namenode -format” da seguinte forma -

$ cd ~ $ hdfs namenode -format

O resultado esperado é o seguinte -

10/24/14 21:30:55 INFO namenode.NameNode: STARTUP_MSG:
/************************************************************
STARTUP_MSG: Starting NameNode
STARTUP_MSG: host = localhost/192.168.1.11
STARTUP_MSG: args = [-format]
STARTUP_MSG: version = 2.4.1
...
...
10/24/14 21:30:56 INFO common.Storage: Storage directory
/home/hadoop/hadoopinfra/hdfs/namenode has been successfully formatted.
10/24/14 21:30:56 INFO namenode.NNStorageRetentionManager: Going to
retain 1 images with txid >= 0
10/24/14 21:30:56 INFO util.ExitUtil: Exiting with status 0
10/24/14 21:30:56 INFO namenode.NameNode: SHUTDOWN_MSG:

/************************************************************
SHUTDOWN_MSG: Shutting down NameNode at localhost/192.168.1.11
************************************************************/

Etapa 2 - Verificar Hadoop dfs

Execute o seguinte comando para iniciar seu sistema de arquivos Hadoop.

$ start-dfs.sh

A saída esperada é a seguinte -

10/24/14 21:37:56
Starting namenodes on [localhost]
localhost: starting namenode, logging to /home/hadoop/hadoop-
2.4.1/logs/hadoop-hadoop-namenode-localhost.out
localhost: starting datanode, logging to /home/hadoop/hadoop-
2.4.1/logs/hadoop-hadoop-datanode-localhost.out
Starting secondary namenodes [0.0.0.0]

Etapa 3 - Verificação do script do Yarn

O seguinte comando é usado para iniciar o script yarn. Executar este comando iniciará seus daemons de yarn.

$ start-yarn.sh

A saída esperada é a seguinte -

starting yarn daemons
starting resourcemanager, logging to /home/hadoop/hadoop-
2.4.1/logs/yarn-hadoop-resourcemanager-localhost.out
localhost: starting node manager, logging to /home/hadoop/hadoop-
2.4.1/logs/yarn-hadoop-nodemanager-localhost.out

Etapa 4 - Acessando o Hadoop no navegador

O número da porta padrão para acessar o Hadoop é 50070. Use a seguinte URL para obter serviços Hadoop em seu navegador.

http://localhost:50070/

A captura de tela a seguir mostra o navegador Hadoop.

Etapa 5 - Verificar todos os aplicativos de um cluster

O número da porta padrão para acessar todos os aplicativos de um cluster é 8088. Use a seguinte URL para usar este serviço.

http://localhost:8088/

A captura de tela a seguir mostra um navegador de cluster Hadoop.

Neste capítulo, daremos uma olhada mais de perto nas classes e seus métodos que estão envolvidos nas operações de programação MapReduce. Manteremos nosso foco principalmente no seguinte -

  • Interface JobContext
  • Classe de Trabalho
  • Classe Mapper
  • Classe de redutor

Interface JobContext

A interface JobContext é a superinterface para todas as classes, que define diferentes trabalhos no MapReduce. Ele fornece uma exibição somente leitura do trabalho fornecido para as tarefas durante sua execução.

A seguir estão as subinterfaces da interface JobContext.

S.No. Descrição da subinterface
1 MapContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT>

Define o contexto que é fornecido ao Mapeador.
2 ReduceContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT>

Define o contexto que é passado para o Redutor.

A classe de trabalho é a classe principal que implementa a interface JobContext.

Classe de Trabalho

A classe Job é a classe mais importante na API MapReduce. Ele permite que o usuário configure o trabalho, envie-o, controle sua execução e consulte o estado. Os métodos definidos só funcionam até que o trabalho seja enviado, depois disso eles lançarão uma IllegalStateException.

Normalmente, o usuário cria o aplicativo, descreve as várias facetas do trabalho e, em seguida, envia o trabalho e monitora seu progresso.

Aqui está um exemplo de como enviar um trabalho -

// Create a new Job
Job job = new Job(new Configuration());
job.setJarByClass(MyJob.class);

// Specify various job-specific parameters
job.setJobName("myjob");
job.setInputPath(new Path("in"));
job.setOutputPath(new Path("out"));

job.setMapperClass(MyJob.MyMapper.class);
job.setReducerClass(MyJob.MyReducer.class);

// Submit the job, then poll for progress until the job is complete
job.waitForCompletion(true);

Construtores

A seguir está o resumo do construtor da classe Job.

S.No Resumo do construtor
1 Job()
2 Job(Configuração conf)
3 Job(Configuração conf, String jobName)

Métodos

Alguns dos métodos importantes da classe Job são os seguintes -

S.No Descrição do Método
1 getJobName()

Nome do trabalho especificado pelo usuário.
2 getJobState()

Retorna o estado atual do Job.
3 isComplete()

Verifica se o trabalho foi concluído ou não.
4 setInputFormatClass()

Define o InputFormat para o trabalho.
5 setJobName(String name)

Define o nome do trabalho especificado pelo usuário.
6 setOutputFormatClass()

Define o formato de saída para o trabalho.
7 setMapperClass(Class)

Define o mapeador para o trabalho.
8 setReducerClass(Class)

Define o redutor para o trabalho.
9 setPartitionerClass(Class)

Define o particionador para o trabalho.
10 setCombinerClass(Class)

Define o combinador para o trabalho.

Classe Mapper

A classe Mapper define o trabalho de mapa. Mapeia pares de valores-chave de entrada para um conjunto de pares de valores-chave intermediários. Mapas são as tarefas individuais que transformam os registros de entrada em registros intermediários. Os registros intermediários transformados não precisam ser do mesmo tipo que os registros de entrada. Um determinado par de entrada pode ser mapeado para zero ou muitos pares de saída.

Método

mapé o método mais proeminente da classe Mapper. A sintaxe é definida abaixo -

map(KEYIN key, VALUEIN value, org.apache.hadoop.mapreduce.Mapper.Context context)

Este método é chamado uma vez para cada par de valores-chave na divisão de entrada.

Classe de redutor

A classe Reducer define o trabalho Reduce no MapReduce. Ele reduz um conjunto de valores intermediários que compartilham uma chave a um conjunto menor de valores. As implementações do redutor podem acessar a configuração de um trabalho por meio do método JobContext.getConfiguration (). Um Redutor possui três fases primárias - Ordem Aleatória, Classificação e Redução.

  • Shuffle - O Redutor copia a saída classificada de cada Mapeador usando HTTP na rede.

  • Sort- A estrutura mescla as entradas do Redutor por chaves (uma vez que mapeadores diferentes podem ter a mesma chave de saída). As fases de embaralhamento e classificação ocorrem simultaneamente, ou seja, enquanto as saídas estão sendo buscadas, elas são mescladas.

  • Reduce - Nesta fase, o método reduzir (Objeto, Iterável, Contexto) é chamado para cada <chave (coleção de valores)> nas entradas classificadas.

Método

reduceé o método mais proeminente da classe Redutor. A sintaxe é definida abaixo -

reduce(KEYIN key, Iterable<VALUEIN> values, org.apache.hadoop.mapreduce.Reducer.Context context)

Este método é chamado uma vez para cada chave na coleção de pares chave-valor.

MapReduce é uma estrutura usada para escrever aplicativos para processar grandes volumes de dados em grandes clusters de hardware comum de maneira confiável. Este capítulo o conduz pela operação de MapReduce na estrutura Hadoop usando Java.

Algoritmo MapReduce

Geralmente o paradigma MapReduce é baseado no envio de programas de redução de mapa para computadores onde os dados reais residem.

  • Durante uma tarefa MapReduce, o Hadoop envia tarefas Map e Reduce para os servidores apropriados no cluster.

  • A estrutura gerencia todos os detalhes da passagem de dados, como emitir tarefas, verificar a conclusão da tarefa e copiar dados em todo o cluster entre os nós.

  • A maior parte da computação ocorre nos nós com dados em discos locais que reduzem o tráfego de rede.

  • Depois de concluir uma determinada tarefa, o cluster coleta e reduz os dados para formar um resultado apropriado e os envia de volta ao servidor Hadoop.

Entradas e saídas (perspectiva Java)

A estrutura MapReduce opera em pares de valor-chave, ou seja, a estrutura visualiza a entrada para a tarefa como um conjunto de pares de valor-chave e produz um conjunto de par de valor-chave como a saída da tarefa, possivelmente de diferentes tipos.

As classes de chave e valor devem ser serializáveis ​​pela estrutura e, portanto, são necessárias para implementar a interface gravável. Além disso, as classes-chave precisam implementar a interface WritableComparable para facilitar a classificação pela estrutura.

Os formatos de entrada e saída de um trabalho MapReduce estão na forma de pares de valores-chave -

(Entrada) <k1, v1> -> mapa -> <k2, v2> -> reduzir -> <k3, v3> (saída).

Entrada Resultado
Mapa <k1, v1> lista (<k2, v2>)
Reduzir <k2, lista (v2)> lista (<k3, v3>)

Implementação MapReduce

A tabela a seguir mostra os dados relativos ao consumo elétrico de uma organização. A tabela inclui o consumo elétrico mensal e a média anual de cinco anos consecutivos.

Jan Fev Mar Abr Maio Junho Jul Agosto Set Out Nov Dez Média
1979 23 23 2 43 24 25 26 26 26 26 25 26 25
1980 26 27 28 28 28 30 31 31 31 30 30 30 29
1981 31 32 32 32 33 34 35 36 36 34 34 34 34
1984 39 38 39 39 39 41 42 43 40 39 38 38 40
1985 38 39 39 39 39 41 41 41 00 40 39 39 45

Precisamos escrever aplicativos para processar os dados de entrada na tabela fornecida para encontrar o ano de uso máximo, o ano de uso mínimo e assim por diante. Esta tarefa é fácil para programadores com uma quantidade finita de registros, pois eles simplesmente escreverão a lógica para produzir a saída necessária e passarão os dados para o aplicativo escrito.

Vamos agora aumentar a escala dos dados de entrada. Suponha que tenhamos que analisar o consumo elétrico de todas as indústrias de grande escala de um determinado estado. Quando escrevemos aplicativos para processar esses dados em massa,

  • Eles levarão muito tempo para serem executados.

  • Haverá tráfego de rede pesado quando movermos os dados da origem para o servidor de rede.

Para resolver esses problemas, temos a estrutura MapReduce.

Dados de entrada

Os dados acima são salvos como sample.txte fornecido como entrada. O arquivo de entrada se parece com o mostrado abaixo.

1979 23 23 2 43 24 25 26 26 26 26 25 26 25
1980 26 27 28 28 28 30 31 31 31 30 30 30 29
1981 31 32 32 32 33 34 35 36 36 34 34 34 34
1984 39 38 39 39 39 41 42 43 40 39 38 38 40
1985 38 39 39 39 39 41 41 41 00 40 39 39 45

Programa Exemplo

O programa a seguir para os dados de amostra usa a estrutura MapReduce.

package hadoop;

import java.util.*;
import java.io.IOException;
import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

public class ProcessUnits
{
   //Mapper class
   public static class E_EMapper extends MapReduceBase implements
   Mapper<LongWritable,  /*Input key Type */
   Text,                   /*Input value Type*/
   Text,                   /*Output key Type*/
   IntWritable>            /*Output value Type*/
   {
      //Map function
      public void map(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException
      {
         String line = value.toString();
         String lasttoken = null;
         StringTokenizer s = new StringTokenizer(line,"\t");
         String year = s.nextToken();
         
         while(s.hasMoreTokens()){
            lasttoken=s.nextToken();
         }
         
         int avgprice = Integer.parseInt(lasttoken);
         output.collect(new Text(year), new IntWritable(avgprice));
      }
   }
   
   //Reducer class
	
   public static class E_EReduce extends MapReduceBase implements
   Reducer< Text, IntWritable, Text, IntWritable >
   {
      //Reduce function
      public void reduce(Text key, Iterator <IntWritable> values, OutputCollector>Text, IntWritable> output, Reporter reporter) throws IOException
      {
         int maxavg=30;
         int val=Integer.MIN_VALUE;
         while (values.hasNext())
         {
            if((val=values.next().get())>maxavg)
            {
               output.collect(key, new IntWritable(val));
            }
         }
      }
   }
	
   //Main function
	
   public static void main(String args[])throws Exception
   {
      JobConf conf = new JobConf(Eleunits.class);
		
      conf.setJobName("max_eletricityunits");
		
      conf.setOutputKeyClass(Text.class);
      conf.setOutputValueClass(IntWritable.class);
		
      conf.setMapperClass(E_EMapper.class);
      conf.setCombinerClass(E_EReduce.class);
      conf.setReducerClass(E_EReduce.class);
		
      conf.setInputFormat(TextInputFormat.class);
      conf.setOutputFormat(TextOutputFormat.class);
		
      FileInputFormat.setInputPaths(conf, new Path(args[0]));
      FileOutputFormat.setOutputPath(conf, new Path(args[1]));
		
      JobClient.runJob(conf);
   }
}

Salve o programa acima em ProcessUnits.java. A compilação e execução do programa são fornecidas abaixo.

Compilação e execução do programa ProcessUnits

Vamos supor que estamos no diretório inicial do usuário Hadoop (por exemplo, / home / hadoop).

Siga as etapas fornecidas a seguir para compilar e executar o programa acima.

Step 1 - Use o seguinte comando para criar um diretório para armazenar as classes java compiladas.

$ mkdir units

Step 2- Baixe Hadoop-core-1.2.1.jar, que é usado para compilar e executar o programa MapReduce. Baixe o jar em mvnrepository.com . Vamos supor que a pasta de download seja / home / hadoop /.

Step 3 - Os seguintes comandos são usados ​​para compilar o ProcessUnits.java programa e para criar um jar para o programa.

$ javac -classpath hadoop-core-1.2.1.jar -d units ProcessUnits.java
$ jar -cvf units.jar -C units/ .

Step 4 - O seguinte comando é usado para criar um diretório de entrada no HDFS.

$HADOOP_HOME/bin/hadoop fs -mkdir input_dir

Step 5 - O seguinte comando é usado para copiar o arquivo de entrada chamado sample.txt no diretório de entrada do HDFS.

$HADOOP_HOME/bin/hadoop fs -put /home/hadoop/sample.txt input_dir

Step 6 - O seguinte comando é usado para verificar os arquivos no diretório de entrada

$HADOOP_HOME/bin/hadoop fs -ls input_dir/

Step 7 - O comando a seguir é usado para executar o aplicativo Eleunit_max, obtendo arquivos de entrada do diretório de entrada.

$HADOOP_HOME/bin/hadoop jar units.jar hadoop.ProcessUnits input_dir output_dir

Espere um pouco até que o arquivo seja executado. Após a execução, a saída contém uma série de divisões de entrada, tarefas de mapa, tarefas de redutor, etc.

INFO mapreduce.Job: Job job_1414748220717_0002
completed successfully
14/10/31 06:02:52
INFO mapreduce.Job: Counters: 49

File System Counters
   
   FILE: Number of bytes read=61
   FILE: Number of bytes written=279400
   FILE: Number of read operations=0
   FILE: Number of large read operations=0
   FILE: Number of write operations=0

   HDFS: Number of bytes read=546
   HDFS: Number of bytes written=40
   HDFS: Number of read operations=9
   HDFS: Number of large read operations=0
   HDFS: Number of write operations=2 Job Counters
   
   Launched map tasks=2
   Launched reduce tasks=1
   Data-local map tasks=2
	
   Total time spent by all maps in occupied slots (ms)=146137
   Total time spent by all reduces in occupied slots (ms)=441
   Total time spent by all map tasks (ms)=14613
   Total time spent by all reduce tasks (ms)=44120
	
   Total vcore-seconds taken by all map tasks=146137
   Total vcore-seconds taken by all reduce tasks=44120
	
   Total megabyte-seconds taken by all map tasks=149644288
   Total megabyte-seconds taken by all reduce tasks=45178880

Map-Reduce Framework
   
   Map input records=5
	
   Map output records=5
   Map output bytes=45
   Map output materialized bytes=67
	
   Input split bytes=208
   Combine input records=5
   Combine output records=5
	
   Reduce input groups=5
   Reduce shuffle bytes=6
   Reduce input records=5
   Reduce output records=5
	
   Spilled Records=10
   Shuffled Maps =2
   Failed Shuffles=0
   Merged Map outputs=2
	
   GC time elapsed (ms)=948
   CPU time spent (ms)=5160
	
   Physical memory (bytes) snapshot=47749120
   Virtual memory (bytes) snapshot=2899349504
	
   Total committed heap usage (bytes)=277684224

File Output Format Counters

   Bytes Written=40

Step 8 - O seguinte comando é usado para verificar os arquivos resultantes na pasta de saída.

$HADOOP_HOME/bin/hadoop fs -ls output_dir/

Step 9 - O seguinte comando é usado para ver a saída em Part-00000Arquivo. Este arquivo é gerado pelo HDFS.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000

A seguir está a saída gerada pelo programa MapReduce -

1981 34
1984 40
1985 45

Step 10 - O seguinte comando é usado para copiar a pasta de saída do HDFS para o sistema de arquivos local.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000/bin/hadoop dfs -get output_dir /home/hadoop

Um particionador funciona como uma condição no processamento de um conjunto de dados de entrada. A fase de partição ocorre após a fase de Mapa e antes da fase de Redução.

O número de particionadores é igual ao número de redutores. Isso significa que um particionador dividirá os dados de acordo com o número de redutores. Portanto, os dados passados ​​de um único particionador são processados ​​por um único Redutor.

Particionador

Um particionador particiona os pares chave-valor de saídas Map intermediárias. Ele particiona os dados usando uma condição definida pelo usuário, que funciona como uma função hash. O número total de partições é igual ao número de tarefas do Redutor para o trabalho. Vamos dar um exemplo para entender como funciona o particionador.

Implementação do Particionador MapReduce

Por uma questão de conveniência, vamos supor que temos uma pequena tabela chamada Funcionário com os seguintes dados. Usaremos esses dados de amostra como nosso conjunto de dados de entrada para demonstrar como o particionador funciona.

Eu iria Nome Era Gênero Salário
1201 gopal 45 Masculino 50.000
1202 manisha 40 Fêmea 50.000
1203 Khalil 34 Masculino 30.000
1204 prasanth 30 Masculino 30.000
1205 Kiran 20 Masculino 40.000
1206 laxmi 25 Fêmea 35.000
1207 bhavya 20 Fêmea 15.000
1208 reshma 19 Fêmea 15.000
1209 Kranthi 22 Masculino 22.000
1210 Satish 24 Masculino 25.000
1211 Krishna 25 Masculino 25.000
1212 Arshad 28 Masculino 20.000
1213 Lavanya 18 Fêmea 8.000

Temos que escrever um aplicativo para processar o conjunto de dados de entrada para encontrar o funcionário mais bem remunerado por gênero em diferentes grupos de idade (por exemplo, abaixo de 20, entre 21 a 30, acima de 30).

Dados de entrada

Os dados acima são salvos como input.txt no diretório “/ home / hadoop / hadoopPartitioner” e fornecido como entrada.

1201 gopal 45 Masculino 50000
1202 manisha 40 Fêmea 51000
1203 khaleel 34 Masculino 30000
1204 prasanth 30 Male 31000
1205 kiran 20 Male 40000
1206 laxmi 25 Female 35000
1207 bhavya 20 Female 15000
1208 reshma 19 Female 14000
1209 kranthi 22 Male 22000
1210 Satish 24 Male 25000
1211 Krishna 25 Male 26000
1212 Arshad 28 Male 20000
1213 lavanya 18 Female 8000

Based on the given input, following is the algorithmic explanation of the program.

Map Tasks

The map task accepts the key-value pairs as input while we have the text data in a text file. The input for this map task is as follows −

Input − The key would be a pattern such as “any special key + filename + line number” (example: key = @input1) and the value would be the data in that line (example: value = 1201 \t gopal \t 45 \t Male \t 50000).

Method − The operation of this map task is as follows −

  • Read the value (record data), which comes as input value from the argument list in a string.

  • Using the split function, separate the gender and store in a string variable.

String[] str = value.toString().split("\t", -3);
String gender=str[3];

  • Send the gender information and the record data value as output key-value pair from the map task to the partition task.

context.write(new Text(gender), new Text(value));

  • Repeat all the above steps for all the records in the text file.

Output − You will get the gender data and the record data value as key-value pairs.

Partitioner Task

The partitioner task accepts the key-value pairs from the map task as its input. Partition implies dividing the data into segments. According to the given conditional criteria of partitions, the input key-value paired data can be divided into three parts based on the age criteria.

Input − The whole data in a collection of key-value pairs.

key = Gender field value in the record.

value = Whole record data value of that gender.

Method − The process of partition logic runs as follows.

  • Read the age field value from the input key-value pair.
String[] str = value.toString().split("\t");
int age = Integer.parseInt(str[2]);

  • Check the age value with the following conditions.

    • Age less than or equal to 20
    • Age Greater than 20 and Less than or equal to 30.
    • Age Greater than 30.
if(age<=20)
{
   return 0;
}
else if(age>20 && age<=30)
{
   return 1 % numReduceTasks;
}
else
{
   return 2 % numReduceTasks;
}

Output − The whole data of key-value pairs are segmented into three collections of key-value pairs. The Reducer works individually on each collection.

Reduce Tasks

The number of partitioner tasks is equal to the number of reducer tasks. Here we have three partitioner tasks and hence we have three Reducer tasks to be executed.

Input − The Reducer will execute three times with different collection of key-value pairs.

key = gender field value in the record.

value = the whole record data of that gender.

Method − The following logic will be applied on each collection.

  • Read the Salary field value of each record.
String [] str = val.toString().split("\t", -3);
Note: str[4] have the salary field value.

  • Check the salary with the max variable. If str[4] is the max salary, then assign str[4] to max, otherwise skip the step.

if(Integer.parseInt(str[4])>max)
{
   max=Integer.parseInt(str[4]);
}

  • Repeat Steps 1 and 2 for each key collection (Male & Female are the key collections). After executing these three steps, you will find one max salary from the Male key collection and one max salary from the Female key collection.

context.write(new Text(key), new IntWritable(max));

Output − Finally, you will get a set of key-value pair data in three collections of different age groups. It contains the max salary from the Male collection and the max salary from the Female collection in each age group respectively.

After executing the Map, the Partitioner, and the Reduce tasks, the three collections of key-value pair data are stored in three different files as the output.

All the three tasks are treated as MapReduce jobs. The following requirements and specifications of these jobs should be specified in the Configurations −

  • Job name
  • Input and Output formats of keys and values
  • Individual classes for Map, Reduce, and Partitioner tasks
Configuration conf = getConf();

//Create Job
Job job = new Job(conf, "topsal");
job.setJarByClass(PartitionerExample.class);

// File Input and Output paths
FileInputFormat.setInputPaths(job, new Path(arg[0]));
FileOutputFormat.setOutputPath(job,new Path(arg[1]));

//Set Mapper class and Output format for key-value pair.
job.setMapperClass(MapClass.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);

//set partitioner statement
job.setPartitionerClass(CaderPartitioner.class);

//Set Reducer class and Input/Output format for key-value pair.
job.setReducerClass(ReduceClass.class);

//Number of Reducer tasks.
job.setNumReduceTasks(3);

//Input and Output format for data
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

Programa Exemplo

O programa a seguir mostra como implementar os particionadores para os critérios fornecidos em um programa MapReduce.

package partitionerexample;

import java.io.*;

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.fs.*;

import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;

import org.apache.hadoop.util.*;

public class PartitionerExample extends Configured implements Tool
{
   //Map class
	
   public static class MapClass extends Mapper<LongWritable,Text,Text,Text>
   {
      public void map(LongWritable key, Text value, Context context)
      {
         try{
            String[] str = value.toString().split("\t", -3);
            String gender=str[3];
            context.write(new Text(gender), new Text(value));
         }
         catch(Exception e)
         {
            System.out.println(e.getMessage());
         }
      }
   }
   
   //Reducer class
	
   public static class ReduceClass extends Reducer<Text,Text,Text,IntWritable>
   {
      public int max = -1;
      public void reduce(Text key, Iterable <Text> values, Context context) throws IOException, InterruptedException
      {
         max = -1;
			
         for (Text val : values)
         {
            String [] str = val.toString().split("\t", -3);
            if(Integer.parseInt(str[4])>max)
            max=Integer.parseInt(str[4]);
         }
			
         context.write(new Text(key), new IntWritable(max));
      }
   }
   
   //Partitioner class
	
   public static class CaderPartitioner extends
   Partitioner < Text, Text >
   {
      @Override
      public int getPartition(Text key, Text value, int numReduceTasks)
      {
         String[] str = value.toString().split("\t");
         int age = Integer.parseInt(str[2]);
         
         if(numReduceTasks == 0)
         {
            return 0;
         }
         
         if(age<=20)
         {
            return 0;
         }
         else if(age>20 && age<=30)
         {
            return 1 % numReduceTasks;
         }
         else
         {
            return 2 % numReduceTasks;
         }
      }
   }
   
   @Override
   public int run(String[] arg) throws Exception
   {
      Configuration conf = getConf();
		
      Job job = new Job(conf, "topsal");
      job.setJarByClass(PartitionerExample.class);
		
      FileInputFormat.setInputPaths(job, new Path(arg[0]));
      FileOutputFormat.setOutputPath(job,new Path(arg[1]));
		
      job.setMapperClass(MapClass.class);
		
      job.setMapOutputKeyClass(Text.class);
      job.setMapOutputValueClass(Text.class);
      
      //set partitioner statement
		
      job.setPartitionerClass(CaderPartitioner.class);
      job.setReducerClass(ReduceClass.class);
      job.setNumReduceTasks(3);
      job.setInputFormatClass(TextInputFormat.class);
		
      job.setOutputFormatClass(TextOutputFormat.class);
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(Text.class);
		
      System.exit(job.waitForCompletion(true)? 0 : 1);
      return 0;
   }
   
   public static void main(String ar[]) throws Exception
   {
      int res = ToolRunner.run(new Configuration(), new PartitionerExample(),ar);
      System.exit(0);
   }
}

Salve o código acima como PartitionerExample.javaem “/ home / hadoop / hadoopPartitioner”. A compilação e execução do programa são fornecidas abaixo.

Compilação e execução

Suponhamos que estamos no diretório inicial do usuário Hadoop (por exemplo, / home / hadoop).

Siga as etapas fornecidas a seguir para compilar e executar o programa acima.

Step 1- Baixe Hadoop-core-1.2.1.jar, que é usado para compilar e executar o programa MapReduce. Você pode baixar o jar em mvnrepository.com .

Suponhamos que a pasta baixada seja “/ home / hadoop / hadoopPartitioner”

Step 2 - Os seguintes comandos são usados ​​para compilar o programa PartitionerExample.java e criando um jar para o programa.

$ javac -classpath hadoop-core-1.2.1.jar -d ProcessUnits.java $ jar -cvf PartitionerExample.jar -C .

Step 3 - Use o seguinte comando para criar um diretório de entrada no HDFS.

$HADOOP_HOME/bin/hadoop fs -mkdir input_dir

Step 4 - Use o seguinte comando para copiar o arquivo de entrada chamado input.txt no diretório de entrada do HDFS.

$HADOOP_HOME/bin/hadoop fs -put /home/hadoop/hadoopPartitioner/input.txt input_dir

Step 5 - Use o seguinte comando para verificar os arquivos no diretório de entrada.

$HADOOP_HOME/bin/hadoop fs -ls input_dir/

Step 6 - Use o seguinte comando para executar o aplicativo Salário superior, obtendo arquivos de entrada do diretório de entrada.

$HADOOP_HOME/bin/hadoop jar PartitionerExample.jar partitionerexample.PartitionerExample input_dir/input.txt output_dir

Espere um pouco até que o arquivo seja executado. Após a execução, a saída contém uma série de divisões de entrada, tarefas de mapa e tarefas de Redutor.

15/02/04 15:19:51 INFO mapreduce.Job: Job job_1423027269044_0021 completed successfully
15/02/04 15:19:52 INFO mapreduce.Job: Counters: 49

File System Counters

   FILE: Number of bytes read=467
   FILE: Number of bytes written=426777
   FILE: Number of read operations=0
   FILE: Number of large read operations=0
   FILE: Number of write operations=0
	
   HDFS: Number of bytes read=480
   HDFS: Number of bytes written=72
   HDFS: Number of read operations=12
   HDFS: Number of large read operations=0
   HDFS: Number of write operations=6
	
Job Counters

   Launched map tasks=1
   Launched reduce tasks=3
	
   Data-local map tasks=1
	
   Total time spent by all maps in occupied slots (ms)=8212
   Total time spent by all reduces in occupied slots (ms)=59858
   Total time spent by all map tasks (ms)=8212
   Total time spent by all reduce tasks (ms)=59858
	
   Total vcore-seconds taken by all map tasks=8212
   Total vcore-seconds taken by all reduce tasks=59858
	
   Total megabyte-seconds taken by all map tasks=8409088
   Total megabyte-seconds taken by all reduce tasks=61294592
	
Map-Reduce Framework

   Map input records=13
   Map output records=13
   Map output bytes=423
   Map output materialized bytes=467
	
   Input split bytes=119
	
   Combine input records=0
   Combine output records=0
	
   Reduce input groups=6
   Reduce shuffle bytes=467
   Reduce input records=13
   Reduce output records=6
	
   Spilled Records=26
   Shuffled Maps =3
   Failed Shuffles=0
   Merged Map outputs=3
   GC time elapsed (ms)=224
   CPU time spent (ms)=3690
	
   Physical memory (bytes) snapshot=553816064
   Virtual memory (bytes) snapshot=3441266688
	
   Total committed heap usage (bytes)=334102528
	
Shuffle Errors

   BAD_ID=0
   CONNECTION=0
   IO_ERROR=0
	
   WRONG_LENGTH=0
   WRONG_MAP=0
   WRONG_REDUCE=0
	
File Input Format Counters

   Bytes Read=361
	
File Output Format Counters

   Bytes Written=72

Step 7 - Use o seguinte comando para verificar os arquivos resultantes na pasta de saída.

$HADOOP_HOME/bin/hadoop fs -ls output_dir/

Você encontrará a saída em três arquivos porque está usando três particionadores e três Redutores em seu programa.

Step 8 - Use o seguinte comando para ver a saída em Part-00000Arquivo. Este arquivo é gerado pelo HDFS.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000

Output in Part-00000

Female   15000
Male     40000

Use o seguinte comando para ver a saída em Part-00001 Arquivo.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00001

Output in Part-00001

Female   35000
Male    31000

Use o seguinte comando para ver a saída em Part-00002 Arquivo.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00002

Output in Part-00002

Female  51000
Male   50000

Um Combiner, também conhecido como semi-reducer, é uma classe opcional que opera aceitando as entradas da classe Map e depois passando os pares de valores-chave de saída para a classe Reducer.

A principal função de um Combiner é resumir os registros de saída do mapa com a mesma tecla. A saída (coleção de valores-chave) do combinador será enviada pela rede para a tarefa do Redutor real como entrada.

Combiner

A classe Combiner é usada entre a classe Map e a classe Reduce para reduzir o volume de transferência de dados entre Map e Reduce. Normalmente, a saída da tarefa de mapa é grande e os dados transferidos para a tarefa de redução são altos.

O seguinte diagrama de tarefas MapReduce mostra a FASE DO COMBINADOR.

Como funciona o combinador?

Aqui está um breve resumo sobre como funciona o MapReduce Combiner -

  • Um combinador não tem uma interface predefinida e deve implementar o método reduz () da interface do Redutor.

  • Um combinador opera em cada chave de saída do mapa. Ela deve ter os mesmos tipos de valores-chave de saída da classe Reducer.

  • Um combinador pode produzir informações resumidas de um grande conjunto de dados porque substitui a saída do mapa original.

Embora o Combiner seja opcional, ele ajuda a segregar os dados em vários grupos para a fase de Redução, o que facilita o processamento.

Implementação do Combinador MapReduce

O exemplo a seguir fornece uma ideia teórica sobre combinadores. Vamos supor que temos o seguinte arquivo de texto de entrada denominadoinput.txt para MapReduce.

What do you mean by Object
What do you know about Java
What is Java Virtual Machine
How Java enabled High Performance

As fases importantes do programa MapReduce com Combiner são discutidas abaixo.

Leitor de registro

Esta é a primeira fase do MapReduce, em que o leitor de registros lê todas as linhas do arquivo de texto de entrada como texto e produz a saída como pares de valores-chave.

Input - Texto linha por linha do arquivo de entrada.

Output- Forma os pares de valores-chave. A seguir está o conjunto de pares de valores-chave esperados.

<1, What do you mean by Object>
<2, What do you know about Java>
<3, What is Java Virtual Machine>
<4, How Java enabled High Performance>

Fase do Mapa

A fase Mapa obtém a entrada do Leitor de registros, processa-a e produz a saída como outro conjunto de pares de chave-valor.

Input - O par de valores-chave a seguir é a entrada obtida do leitor de registros.

<1, What do you mean by Object>
<2, What do you know about Java>
<3, What is Java Virtual Machine>
<4, How Java enabled High Performance>

A fase de Mapa lê cada par de valor-chave, divide cada palavra do valor usando StringTokenizer, trata cada palavra como chave e a contagem dessa palavra como valor. O fragmento de código a seguir mostra a classe Mapper e a função de mapa.

public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>
{
   private final static IntWritable one = new IntWritable(1);
   private Text word = new Text();
   
   public void map(Object key, Text value, Context context) throws IOException, InterruptedException 
   {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) 
      {
         word.set(itr.nextToken());
         context.write(word, one);
      }
   }
}

Output - A saída esperada é a seguinte -

<What,1> <do,1> <you,1> <mean,1> <by,1> <Object,1>
<What,1> <do,1> <you,1> <know,1> <about,1> <Java,1>
<What,1> <is,1> <Java,1> <Virtual,1> <Machine,1>
<How,1> <Java,1> <enabled,1> <High,1> <Performance,1>

Fase Combinador

A fase Combiner pega cada par de valor-chave da fase de Mapa, processa e produz a saída como key-value collection pares.

Input - O seguinte par de valores-chave é a entrada obtida da fase de Mapa.

<What,1> <do,1> <you,1> <mean,1> <by,1> <Object,1>
<What,1> <do,1> <you,1> <know,1> <about,1> <Java,1>
<What,1> <is,1> <Java,1> <Virtual,1> <Machine,1>
<How,1> <Java,1> <enabled,1> <High,1> <Performance,1>

A fase Combiner lê cada par de valor-chave, combina as palavras comuns como chave e os valores como coleção. Normalmente, o código e a operação de um Combiner são semelhantes aos de um Redutor. A seguir está o trecho de código para a declaração das classes Mapper, Combiner e Reducer.

job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);

Output - A saída esperada é a seguinte -

<What,1,1,1> <do,1,1> <you,1,1> <mean,1> <by,1> <Object,1>
<know,1> <about,1> <Java,1,1,1>
<is,1> <Virtual,1> <Machine,1>
<How,1> <enabled,1> <High,1> <Performance,1>

Fase Redutora

A fase Redutor pega cada par de coleção de valor-chave da fase Combiner, processa-o e passa a saída como pares de valor-chave. Observe que a funcionalidade do Combiner é igual à do Redutor.

Input - O seguinte par de valores-chave é a entrada retirada da fase Combiner.

<What,1,1,1> <do,1,1> <you,1,1> <mean,1> <by,1> <Object,1>
<know,1> <about,1> <Java,1,1,1>
<is,1> <Virtual,1> <Machine,1>
<How,1> <enabled,1> <High,1> <Performance,1>

A fase Redutor lê cada par de valor-chave. A seguir está o trecho de código para o Combiner.

public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> 
{
   private IntWritable result = new IntWritable();
   
   public void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException 
   {
      int sum = 0;
      for (IntWritable val : values) 
      {
         sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
   }
}

Output - A saída esperada da fase do Redutor é a seguinte -

<What,3> <do,2> <you,2> <mean,1> <by,1> <Object,1>
<know,1> <about,1> <Java,3>
<is,1> <Virtual,1> <Machine,1>
<How,1> <enabled,1> <High,1> <Performance,1>

Gravador

Esta é a última fase do MapReduce, onde o gravador grava cada par de valores-chave da fase Redutor e envia a saída como texto.

Input - Cada par de valor-chave da fase do Redutor junto com o formato de saída.

Output- Fornece os pares de valores-chave em formato de texto. A seguir está a saída esperada.

What           3
do             2
you            2
mean           1
by             1
Object         1
know           1
about          1
Java           3
is             1
Virtual        1
Machine        1
How            1
enabled        1
High           1
Performance    1

Programa Exemplo

O seguinte bloco de código conta o número de palavras em um programa.

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {
   public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>
   {
      private final static IntWritable one = new IntWritable(1);
      private Text word = new Text();
      
      public void map(Object key, Text value, Context context) throws IOException, InterruptedException 
      {
         StringTokenizer itr = new StringTokenizer(value.toString());
         while (itr.hasMoreTokens()) 
         {
            word.set(itr.nextToken());
            context.write(word, one);
         }
      }
   }
   
   public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> 
   {
      private IntWritable result = new IntWritable();
      public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException 
      {
         int sum = 0;
         for (IntWritable val : values) 
         {
            sum += val.get();
         }
         result.set(sum);
         context.write(key, result);
      }
   }
   
   public static void main(String[] args) throws Exception 
   {
      Configuration conf = new Configuration();
      Job job = Job.getInstance(conf, "word count");
		
      job.setJarByClass(WordCount.class);
      job.setMapperClass(TokenizerMapper.class);
      job.setCombinerClass(IntSumReducer.class);
      job.setReducerClass(IntSumReducer.class);
		
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(IntWritable.class);
		
      FileInputFormat.addInputPath(job, new Path(args[0]));
      FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
      System.exit(job.waitForCompletion(true) ? 0 : 1);
   }
}

Salve o programa acima como WordCount.java. A compilação e execução do programa são fornecidas abaixo.

Compilação e execução

Vamos supor que estejamos no diretório inicial do usuário Hadoop (por exemplo, / home / hadoop).

Siga as etapas fornecidas a seguir para compilar e executar o programa acima.

Step 1 - Use o seguinte comando para criar um diretório para armazenar as classes java compiladas.

$ mkdir units

Step 2- Baixe Hadoop-core-1.2.1.jar, que é usado para compilar e executar o programa MapReduce. Você pode baixar o jar em mvnrepository.com .

Vamos supor que a pasta baixada seja / home / hadoop /.

Step 3 - Use os seguintes comandos para compilar o WordCount.java programa e para criar um jar para o programa.

$ javac -classpath hadoop-core-1.2.1.jar -d units WordCount.java
$ jar -cvf units.jar -C units/ .

Step 4 - Use o seguinte comando para criar um diretório de entrada no HDFS.

$HADOOP_HOME/bin/hadoop fs -mkdir input_dir

Step 5 - Use o seguinte comando para copiar o arquivo de entrada chamado input.txt no diretório de entrada do HDFS.

$HADOOP_HOME/bin/hadoop fs -put /home/hadoop/input.txt input_dir

Step 6 - Use o seguinte comando para verificar os arquivos no diretório de entrada.

$HADOOP_HOME/bin/hadoop fs -ls input_dir/

Step 7 - Use o seguinte comando para executar o aplicativo de contagem de palavras, obtendo arquivos de entrada do diretório de entrada.

$HADOOP_HOME/bin/hadoop jar units.jar hadoop.ProcessUnits input_dir output_dir

Espere um pouco até que o arquivo seja executado. Após a execução, a saída contém várias divisões de entrada, tarefas de mapa e tarefas de redutor.

Step 8 - Use o seguinte comando para verificar os arquivos resultantes na pasta de saída.

$HADOOP_HOME/bin/hadoop fs -ls output_dir/

Step 9 - Use o seguinte comando para ver a saída em Part-00000Arquivo. Este arquivo é gerado pelo HDFS.

$HADOOP_HOME/bin/hadoop fs -cat output_dir/part-00000

A seguir está a saída gerada pelo programa MapReduce.

What           3
do             2
you            2
mean           1
by             1
Object         1
know           1
about          1
Java           3
is             1
Virtual        1
Machine        1
How            1
enabled        1
High           1
Performance    1

Este capítulo explica a administração do Hadoop, que inclui a administração de HDFS e MapReduce.

  • A administração do HDFS inclui o monitoramento da estrutura de arquivos HDFS, locais e arquivos atualizados.

  • A administração do MapReduce inclui o monitoramento da lista de aplicativos, configuração de nós, status do aplicativo, etc.

Monitoramento HDFS

HDFS (Hadoop Distributed File System) contém os diretórios do usuário, arquivos de entrada e arquivos de saída. Use os comandos MapReduce,put e get, para armazenar e recuperar.

Após iniciar a estrutura do Hadoop (daemons) passando o comando “start-all.sh” em “/ $ HADOOP_HOME / sbin”, passe a seguinte URL para o navegador “http: // localhost: 50070”. Você deverá ver a seguinte tela em seu navegador.

A captura de tela a seguir mostra como navegar no HDFS de navegação.

A captura de tela a seguir mostra a estrutura de arquivos do HDFS. Mostra os arquivos no diretório “/ user / hadoop”.

A captura de tela a seguir mostra as informações do Datanode em um cluster. Aqui você pode encontrar um nó com suas configurações e capacidades.

MapReduce Job Monitoring

Um aplicativo MapReduce é uma coleção de trabalhos (trabalho de mapa, combinador, particionador e trabalho de redução). É obrigatório monitorar e manter o seguinte -

  • Configuração do datanode onde a aplicação é adequada.
  • O número de datanodes e recursos usados ​​por aplicativo.

Para monitorar todas essas coisas, é imperativo que tenhamos uma interface de usuário. Depois de iniciar a estrutura do Hadoop passando o comando “start-all.sh” em “/ $ HADOOP_HOME / sbin”, passe a seguinte URL para o navegador “http: // localhost: 8080”. Você deverá ver a seguinte tela em seu navegador.

Na captura de tela acima, o ponteiro da mão está no ID do aplicativo. Basta clicar nele para encontrar a seguinte tela em seu navegador. Ele descreve o seguinte -

  • Em qual usuário o aplicativo atual está sendo executado

  • O nome do aplicativo

  • Tipo desse aplicativo

  • Status atual, status final

  • Tempo de início do aplicativo, decorrido (tempo concluído), se estiver concluído no momento do monitoramento

  • O histórico deste aplicativo, ou seja, informações de registro

  • E, por fim, as informações do nó, ou seja, os nós que participaram da execução da aplicação.

A captura de tela a seguir mostra os detalhes de um determinado aplicativo -

A captura de tela a seguir descreve as informações dos nós em execução no momento. Aqui, a captura de tela contém apenas um nó. Um ponteiro mostra o endereço do host local do nó em execução.

ja/tutorial
es/tutorial
de/tutorial
fr/tutorial
th/tutorial
pt/tutorial
ru/tutorial
vi/tutorial
it/tutorial
ko/tutorial
tr/tutorial
id/tutorial
pl/tutorial
hi/tutorial
japost
espost
depost
frpost
thpost
ptpost
rupost
vipost
itpost
kopost
trpost
idpost
plpost
hipost

© Copyright 2021 - 2024 | All Rights Reserved