Dominando o Particionamento Hadoop para Processamento de Dados Eficiente

Introdução

Na antiga Olimpíada Grega, atletas de toda a terra se reuniam para demonstrar sua proeza e competir em vários eventos atléticos. Um desses atletas, Alexios, treinou incansavelmente para os próximos jogos, determinado a trazer glória para sua cidade-estado.

O objetivo era classificar e organizar os participantes em diferentes grupos com base em seus eventos, garantindo uma competição justa e eficiente. No entanto, com centenas de atletas competindo pela glória, a tarefa de dividi-los em seus respectivos eventos era assustadora.

Implementar o Mapper

Nesta etapa, criaremos uma classe Mapper que lê os dados de entrada e gera pares chave-valor para o Partitioner processar.

Primeiro, altere o usuário para hadoop e, em seguida, mude para o diretório home do usuário hadoop:

su - hadoop

Em seguida, crie um arquivo Java para a classe Mapper:

touch /home/hadoop/OlympicMapper.java

Adicione o seguinte código ao arquivo OlympicMapper.java:

import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class OlympicMapper extends Mapper<LongWritable, Text, Text, Text> {

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] fields = value.toString().split(",");
        String athlete = fields[0];
        String event = fields[1];
        context.write(new Text(event), new Text(athlete));
    }
}

Na classe OlympicMapper, definimos a chave de entrada como LongWritable (representando o deslocamento da linha) e o valor de entrada como Text (representando uma linha de texto do arquivo de entrada). A chave de saída é um objeto Text representando o evento, e o valor de saída é um objeto Text representando o nome do atleta.

O método map divide cada linha de dados de entrada pelo delimitador vírgula, extrai o nome do atleta e o evento, e emite um par chave-valor com o evento como chave e o nome do atleta como valor.

Implementar o Partitioner

Nesta etapa, criaremos uma classe Partitioner personalizada que particiona os pares chave-valor com base no evento.

Primeiro, crie um arquivo Java para a classe Partitioner:

touch /home/hadoop/OlympicPartitioner.java

Em seguida, adicione o seguinte código ao arquivo OlympicPartitioner.java:

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

public class OlympicPartitioner extends Partitioner<Text, Text> {

    @Override
    public int getPartition(Text key, Text value, int numPartitions) {
        return Math.abs(key.hashCode() % numPartitions);
    }
}

A classe OlympicPartitioner estende a classe Partitioner fornecida pelo Hadoop. Ela substitui o método getPartition, que recebe a chave, o valor e o número de partições como entrada.

O método getPartition calcula um código hash para o evento (chave) e retorna o número da partição, pegando o valor absoluto do código hash módulo o número de partições. Isso garante que todos os registros com o mesmo evento sejam enviados para a mesma partição para processamento pelo Reducer.

Implementar o Reducer

Nesta etapa, criaremos uma classe Reducer que processa os dados particionados e gera a saída final.

Primeiro, crie um arquivo Java para a classe Reducer:

touch /home/hadoop/OlympicReducer.java

Em seguida, adicione o seguinte código ao arquivo OlympicReducer.java:

import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class OlympicReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        StringBuilder athletes = new StringBuilder();
        for (Text value : values) {
            athletes.append(value.toString()).append(",");
        }
        if (athletes.length() > 0) {
            athletes.deleteCharAt(athletes.length() - 1);
        }
        context.write(key, new Text(athletes.toString()));
    }
}

A classe OlympicReducer estende a classe Reducer fornecida pelo Hadoop. Ela define a chave de entrada como Text (representando o evento), o valor de entrada como Text (representando o nome do atleta) e a chave e o valor de saída como objetos Text.

O método reduce é chamado para cada chave de evento única, com um iterador sobre os nomes dos atletas associados a esse evento. Ele constrói uma lista separada por vírgulas de atletas para cada evento e emite um par chave-valor com o evento como chave e a lista de atletas como valor.

Escrever o Driver

Nesta etapa, criaremos uma classe Driver que une as classes Mapper, Partitioner e Reducer e executa o trabalho MapReduce.

Primeiro, crie um arquivo Java para a classe Driver:

touch /home/hadoop/OlympicDriver.java

Em seguida, adicione o seguinte código ao arquivo OlympicDriver.java:

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class OlympicDriver {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "Olympic Partitioner");

        job.setJarByClass(OlympicDriver.class);
        job.setMapperClass(OlympicMapper.class);
        job.setPartitionerClass(OlympicPartitioner.class);
        job.setReducerClass(OlympicReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

A classe OlympicDriver é o ponto de entrada para o trabalho MapReduce. Ela configura a configuração do trabalho, especifica as classes Mapper, Partitioner e Reducer e configura os caminhos de entrada e saída.

No método main, criamos um novo objeto Configuration e uma instância Job com o nome do trabalho "Olympic Partitioner". Definimos as classes Mapper, Partitioner e Reducer usando os métodos setter correspondentes.

Também definimos as classes de chave e valor de saída para Text. Os caminhos de entrada e saída são especificados usando argumentos de linha de comando passados para o driver.

Finalmente, chamamos o método waitForCompletion na instância Job para executar o trabalho MapReduce e sair com um código de status apropriado (0 para sucesso, 1 para falha).

Para executar o trabalho, você precisa compilar as classes Java e criar um arquivo jar. Em seguida, você pode executar o arquivo jar usando o seguinte comando:

javac -source 8 -target 8 -classpath "/home/hadoop/:/home/hadoop/hadoop/share/hadoop/common/hadoop-common-3.3.6.jar:/home/hadoop/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-core-3.3.6.jar:/home/hadoop/hadoop/share/hadoop/common/lib/*" -d /home/hadoop /home/hadoop/OlympicMapper.java /home/hadoop/OlympicPartitioner.java /home/hadoop/OlympicReducer.java /home/hadoop/OlympicDriver.java
jar cvf olympic.jar *.class
hadoop jar olympic.jar OlympicDriver /input /output

Finalmente, podemos verificar os resultados executando o seguinte comando:

hadoop fs -cat /output/*

Exemplo de saída:

Event_1 Athlete_17,Athlete_18,Athlete_79,Athlete_71,Athlete_77,Athlete_75,Athlete_19,Athlete_24,Athlete_31,Athlete_32,Athlete_39,Athlete_89,Athlete_88,Athlete_87,Athlete_100,Athlete_13,Athlete_52,Athlete_53,Athlete_58
Event_2 Athlete_1,Athlete_97,Athlete_96,Athlete_85,Athlete_81,Athlete_80,Athlete_72,Athlete_68,Athlete_64,Athlete_61,Athlete_54,Athlete_48,Athlete_47,Athlete_43,Athlete_28,Athlete_23,Athlete_21,Athlete_15,Athlete_12,Athlete_3
Event_3 Athlete_11,Athlete_55,Athlete_8,Athlete_46,Athlete_42,Athlete_41,Athlete_40,Athlete_38,Athlete_33,Athlete_92,Athlete_29,Athlete_27,Athlete_25,Athlete_93,Athlete_22,Athlete_20,Athlete_98,Athlete_14,Athlete_69,Athlete_99,Athlete_66,Athlete_65
Event_4 Athlete_90,Athlete_50,Athlete_37,Athlete_36,Athlete_91,Athlete_74,Athlete_73,Athlete_63,Athlete_26,Athlete_78,Athlete_5,Athlete_62,Athlete_60,Athlete_59,Athlete_82,Athlete_4,Athlete_51,Athlete_86,Athlete_2,Athlete_94,Athlete_7,Athlete_95
Event_5 Athlete_34,Athlete_76,Athlete_57,Athlete_56,Athlete_30,Athlete_16,Athlete_6,Athlete_10,Athlete_83,Athlete_84,Athlete_70,Athlete_45,Athlete_44,Athlete_49,Athlete_9,Athlete_67,Athlete_35

Resumo

Neste laboratório, exploramos o conceito de Hadoop Shuffle Partitioner, projetando um cenário inspirado na antiga Olimpíada Grega. Implementamos uma classe Mapper para ler dados de entrada e gerar pares chave-valor, uma classe Partitioner personalizada para particionar os dados com base no evento e uma classe Reducer para processar os dados particionados e gerar a saída final.

Por meio deste laboratório, ganhei experiência prática com o modelo de programação MapReduce e aprendi como aproveitar a classe Partitioner para distribuir dados de forma eficiente entre as partições. O cenário da antiga Olimpíada Grega forneceu um contexto envolvente para entender as aplicações práticas do Shuffle Partitioner em um ambiente do mundo real.

Particionamento na Olimpíada Hadoop