Junção e contador do Hadoop MapReduce com exemplo

Por: Prafulla Ranadive Prafulla Ranadive
Hours Atualização do

⚡ Resumo Inteligente

As junções MapReduce combinam dois grandes conjuntos de dados com base em uma chave compartilhada, seja dentro do mapeador ou dentro do redutor, enquanto os contadores MapReduce coletam estatísticas sobre a tarefa para que os registros com problemas possam ser medidos em vez de apenas estimados.

  • 🔘 Participe do básico: O menor dos dois conjuntos de dados é distribuído para cada nó de dados e usado como referência para consulta.
  • ☑️ Junção pelo lado do mapa: Requer que cada entrada seja particionada, dividida igualmente e ordenada pela chave de junção antes da execução da função de mapeamento.
  • ✅ Junção do lado redutor: Não necessita de particionamento, pois cada tupla que compartilha uma chave de junção acaba no mesmo redutor.
  • 🧪 Exemplo resolvido: Os arquivos DeptName.txt e DeptStrength.txt são copiados para o HDFS e unidos pelo parâmetro Dept_ID em um arquivo JAR empacotado.
  • 🛠️ Tipos de contador: Cinco grupos de contadores integrados são fornecidos com cada tarefa, e os contadores definidos pelo usuário são declarados como um... Java enumeração.
  • ⚠️ Contra-uso: Ao incrementar um contador para cada registro ausente ou inválido, os problemas de qualidade dos dados se transformam em um número no relatório de tarefas.
Saiba Mais

Tutorial de junção e contador do Hadoop MapReduce com um exemplo prático.

O que é o comando Join no MapReduce?

A operação de junção do MapReduce é usada para combinar dois grandes conjuntos de dados. No entanto, esse processo envolve escrever muito código para realizar a operação de junção propriamente dita. A junção de dois conjuntos de dados começa com a comparação do tamanho de cada um. Se um conjunto de dados for menor que o outro, o conjunto menor será distribuído entre todos os nós de dados do cluster.

Assim que você se juntar a nós MapaReduzir Se os dados forem distribuídos, o Mapper ou o Reducer usa o conjunto de dados menor para realizar uma busca por registros correspondentes no conjunto de dados maior e, em seguida, combina esses registros para formar os registros de saída.

Tipos de Junção

Dependendo do local onde a junção é realizada, as junções no Hadoop são classificadas em dois tipos.

  1. Junção lateral do mapa — Quando a junção é realizada pelo mapeador, ela é chamada de junção do lado do mapa. Nesse tipo, a junção é realizada antes que os dados sejam efetivamente consumidos pela função de mapeamento. É obrigatório que a entrada para cada mapeamento esteja na forma de uma partição e em ordem crescente. Além disso, deve haver um número igual de partições e elas devem estar ordenadas pela chave de junção.
  2. Junção lateral reduzida — Quando a junção é realizada pelo redutor, ela é chamada de junção do lado do redutor. Nesse tipo de junção, não é necessário que o conjunto de dados esteja em um formato estruturado (ou particionado). Aqui, o processamento do lado do mapa gera a chave de junção e as tuplas correspondentes de ambas as tabelas. Como resultado desse processamento, todas as tuplas com a mesma chave de junção são enviadas para o mesmo redutor, que então une os registros com a mesma chave de junção.

Um fluxo geral do processo de junções no Hadoop é representado no diagrama abaixo.

Diagrama de fluxo de processo comparando uma junção do lado do mapa com uma junção do lado da redução no Hadoop.
Tipos de junções no Hadoop MapReduce

Com as duas variantes esclarecidas, a próxima seção aborda uma junção do lado de redução em dois pequenos arquivos departamentais.

Como juntar dois conjuntos de dados: exemplo de MapReduce

Existem dois conjuntos de dados em dois arquivos diferentes (mostrados abaixo). A chave Dept_ID é comum a ambos os arquivos. O objetivo é usar o MapReduce Join para combinar esses arquivos.

Primeiro arquivo de entrada listando os IDs dos departamentos juntamente com os nomes dos departamentos.

1 arquivo
O segundo arquivo de entrada lista os IDs dos departamentos juntamente com os valores de efetivo de cada departamento.

2 arquivo

Entrada: O conjunto de dados de entrada consiste em um arquivo txt, DeptName.txt e DeptStrength.txt.

Baixe os arquivos de entrada daqui

Certifique-se de ter Hadoop instalado. Antes de iniciar o processo do exemplo MapReduce Join, altere o usuário para 'hduser' (ID usado durante a configuração do Hadoop; você pode usar o mesmo ID de usuário utilizado na configuração do Hadoop).

su - hduser_

O prompt muda para a conta Hadoop, conforme mostrado abaixo.

Terminal após alternar para a conta hduser com o comando su

Passo 1) Copie o arquivo zip para o local de sua escolha

O arquivo MapReduceJoin baixado foi colocado no diretório de trabalho escolhido.

Passo 2) Descompacte o arquivo Zip

sudo tar -xvf MapReduceJoin.tar.gz

O extracOs nomes dos arquivos Ted vão rolando na tela enquanto o tar descompacta o arquivo.

Listagem de arquivos no console extracted de MapReduceJoin.tar.gz

Passo 3) Vá para o diretório MapReduceJoin/

cd MapReduceJoin/

Prompt do shell após entrar no diretório MapReduceJoin

Passo 4) Inicie o Hadoop

$HADOOP_HOME/sbin/start-dfs.sh
$HADOOP_HOME/sbin/start-yarn.sh

Ambos os scripts imprimem os daemons que iniciam.

Mensagens de inicialização dos scripts daemon do HDFS e do YARN

Passo 5) DeptStrength.txt e DeptName.txt são os arquivos de entrada usados ​​para este programa de exemplo MapReduce Join.

Esses arquivos precisam ser copiados para HDFS usando o comando abaixo-

$HADOOP_HOME/bin/hdfs dfs -copyFromLocal DeptStrength.txt DeptName.txt /

Ambos os arquivos de texto de entrada foram copiados para o diretório raiz do HDFS.

ARTIGOS RELACIONADOS

Passo 6) Execute o programa usando o comando abaixo-

$HADOOP_HOME/bin/hadoop jar MapReduceJoin.jar MapReduceJoin/JoinDriver/DeptStrength.txt /DeptName.txt /output_mapreducejoin

O comando é reproduzido primeiro e, em seguida, a tarefa relata seu progresso no console.

Linha de comando para executar o arquivo jar do MapReduceJoin.

Saída do console tracrei o progresso do MapReduce junte-se ao trabalho

Passo 7) Após a execução, o arquivo de saída (chamado 'part-00000') será armazenado no diretório /output_mapreducejoin no HDFS.

Os resultados podem ser vistos usando a interface da linha de comando

$HADOOP_HOME/bin/hdfs dfs -cat /output_mapreducejoin/part-00000

Registros do departamento unificados impressos do HDFS pelo comando cat

Os resultados também podem ser vistos através de uma interface web como-

Página inicial da interface web do Hadoop usada para acessar o navegador do sistema de arquivos.

Agora selecione 'Procurar no sistema de arquivos' e navegue até /output_mapreducejoin

Navegando pelo sistema de arquivos HDFS até o diretório output_mapreducejoin.

Abrir parte-r-00000

Selecionando o arquivo de saída part-r-00000 na visualização do navegador.

Os resultados são mostrados

Linhas com o nome do departamento participante e o número de funcionários exibidas no navegador.

OBSERVAÇÃO: Observe que antes de executar este programa pela próxima vez, você precisará excluir o diretório de saída /output_mapreducejoin

$HADOOP_HOME/bin/hdfs dfs -rm -r /output_mapreducejoin

A alternativa é usar um nome diferente para o diretório de saída.

As junções informam qual é o formato dos dados. Os contadores, que serão abordados a seguir, informam como o trabalho que os gerou se comportou.

O que é contador no MapReduce?

Um contador no MapReduce é um mecanismo usado para coletar e medir informações estatísticas sobre trabalhos e eventos do MapReduce. Os contadores mantêm o track contém diversas estatísticas de tarefas no MapReduce, como o número de operações realizadas e o progresso de cada operação. Os contadores são usados ​​para diagnóstico de problemas no MapReduce.

Os contadores Hadoop são semelhantes a colocar uma mensagem de log no código para um mapa ou redução. Esta informação pode ser útil para diagnosticar um problema no processamento do trabalho MapReduce.

Normalmente, esses contadores no Hadoop são definidos em um programa (map ou reduce) e são incrementados durante a execução quando ocorre um evento ou condição específica (a esse contador). Uma ótima aplicação dos contadores do Hadoop é... track registros válidos e inválidos de um conjunto de dados de entrada.

Tipos de contadores MapReduce

Existem basicamente dois tipos de contadores MapReduce.

  1. Contadores integrados do Hadoop: Existem alguns contadores Hadoop integrados que existem por trabalho. Abaixo estão os grupos de contadores integrados-
    • Contadores de tarefas MapReduce — Coleta informações específicas da tarefa (por exemplo, número de registros de entrada) durante seu tempo de execução.
    • Contadores do sistema de arquivos — Coleta informações como o número de bytes lidos ou gravados por uma tarefa.
    • Contadores FileInputFormat — Coleta informações de um número de bytes lidos através do FileInputFormat.
    • Contadores FileOutputFormat — Coleta informações sobre o número de bytes gravados através do FileOutputFormat.
    • Contadores de Emprego — Esses contadores registram estatísticas gerais do trabalho, como o número de tarefas iniciadas para um trabalho.
  2. Contadores definidos pelo usuário: Além dos contadores integrados, um usuário pode definir seus próprios contadores usando funcionalidades semelhantes oferecidas por linguagens de programação. Por exemplo, em JavaUm 'enum' é usado para definir contadores definidos pelo usuário.

💡 Nota da versão: Os contadores de empregos eram mantidos pelo Job.Tracker sob MRv1. No YARN, essa função pertence ao ApplicationMaster do MapReduce, portanto os nomes dos contadores permanecem, mas o componente que os reporta foi alterado.

Um trabalho não pode declarar um número ilimitado de contadores. mapreduce.job.counters.max A configuração limita o total por tarefa a 120 por padrão, e uma tarefa que declare mais do que isso falha com um erro. LimitExceededExceptionPortanto, os contadores são projetados para um conjunto de sinais agregados, em vez de contagens por tecla.

Exemplo de contadores

Um exemplo de classe MapClass com contadores para contar o número de valores ausentes e inválidos. Arquivo de dados de entrada usado neste tutorial: Nosso conjunto de dados de entrada é um arquivo CSV, SalesJan2009.csv.

public static class MapClass
            extends MapReduceBase
            implements Mapper<LongWritable, Text, Text, Text>
{
    static enum SalesCounters { MISSING, INVALID };
    public void map ( LongWritable key, Text value,
                 OutputCollector<Text, Text> output,
                 Reporter reporter) throws IOException
    {
        
        //Input string is split using ',' and stored in 'fields' array
        String fields[] = value.toString().split(",", -20);
        //Value at 4th index is country. It is stored in 'country' variable
        String country = fields[4];
        
        //Value at 8th index is sales data. It is stored in 'sales' variable
        String sales = fields[8];
      
        if (country.length() == 0) {
            reporter.incrCounter(SalesCounters.MISSING, 1);
        } else if (sales.startsWith("\"")) {
            reporter.incrCounter(SalesCounters.INVALID, 1);
        } else {
            output.collect(new Text(country), new Text(sales + ",1"));
        }
    }
}

O trecho de código acima mostra um exemplo de implementação de contadores no Hadoop MapReduce.

Aqui, Contadores de vendas é um contador definido usando 'enumerarÉ utilizado para contar registros de entrada AUSENTES e INVÁLIDOS.

No trecho de código, se 'paísSe o campo tiver comprimento zero, seu valor está ausente e, portanto, o contador correspondente, SalesCounters.MISSING, é incrementado.

Em seguida, se 'vendasSe o campo começar com um “, o registro será considerado INVÁLIDO. Isso é indicado pelo incremento do contador SalesCounters.INVALID.

💡 Nota sobre a API: O trecho acima usa o original org.apache.hadoop.mapred API, onde MapReduceBase, Mapper interface, OutputCollector e Reporter aparecem separadamente. O código atual foi escrito contra org.apache.hadoop.mapreduce, onde um único Context substitui o coletor e o repórter, e um contador é incrementado com context.getCounter(SalesCounters.MISSING).increment(1)O conceito de contador é idêntico em ambos os casos.

Perguntas Frequentes

Escolha a variante de mapeamento quando um dos lados for pequeno o suficiente para caber na memória de cada nó, pois ela evita completamente o embaralhamento. Escolha a variante de redução quando ambos os lados forem grandes ou não ordenados e aceite o custo adicional de rede.

Os modelos aprendem com o histórico de tarefas para prever o tempo de execução, recomendar tamanhos de divisão e número de redutores, e detectar distorções a partir dos valores dos contadores. Eles também sinalizam tarefas cujos contadores de registros derramados ou tarefas com falha estejam fora da faixa normal para aquele pipeline.

O Copilot gera esqueletos plausíveis de mapeadores e redutores, mas mistura livremente o pacote antigo mapred com o pacote mais recente mapreduce em uma única classe, o que não compilará. Corrija as importações e as assinaturas dos métodos antes de confiar na lógica.

É o mecanismo que envia o arquivo menor para cada nó antes do início das tarefas. Cada mapeador então carrega essa cópia em um mapa hash e busca correspondências localmente, o que torna possível uma junção no lado do mapeador.

Eles são impressos no resumo do console quando o trabalho é concluído, expostos no histórico de trabalhos e nas páginas da web do gerenciador de recursos, e podem ser lidos programaticamente a partir do objeto de trabalho, para que um driver possa fazer uma asserção sobre eles e falhar em uma execução com falha.

Um único objeto Context. Ele carrega o trabalho que OutputCollector e Reporter costumavam dividir entre si, de modo que a saída é escrita e os contadores são incrementados através do mesmo identificador passado para o método map.

Para a maioria dos trabalhos de reportagem, não. A junção HiveQL A compilação resulta no mesmo padrão de embaralhamento e mesclagem em poucas linhas. Tarefas escritas manualmente valem a pena quando a lógica de mesclagem não se encaixa em uma cláusula SQL.

O Hadoop se recusa a gravar em um diretório de saída que já existe, o que protege os resultados finais contra sobrescrita. Exclua o diretório recursivamente primeiro ou passe um caminho de saída diferente na próxima execução.

Resuma esta postagem com: