Particionamiento de la Olimpiada de Hadoop

Introducción

En la antigua Olimpiada griega, atletas de todo el territorio se reunían para mostrar su destreza y competir en varios eventos atléticos. Uno de estos atletas, Alexios, había entrenado incansablemente para los próximos juegos, decidido a traer gloria a su ciudad-estado.

El objetivo era clasificar y organizar a los participantes en diferentes grupos según sus eventos, lo que aseguraba una competición justa y eficiente. Sin embargo, con cientos de atletas compitiendo por la gloria, la tarea de dividirlos en sus respectivos eventos era desafiante.

Skills Graph

Implementar el Mapper

En este paso, crearemos una clase Mapper que lea los datos de entrada y genere pares clave-valor para que el Particionador los procese.

Primero, cambie el usuario a hadoop y luego cambie al directorio home del usuario hadoop:

su - hadoop

Luego, cree un archivo Java para la clase Mapper:

touch /home/hadoop/OlympicMapper.java

Agregue el siguiente código al archivo 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));
    }
}

En la clase OlympicMapper, definimos la clave de entrada como LongWritable (que representa el desplazamiento de línea) y el valor de entrada como Text (que representa una línea de texto del archivo de entrada). La clave de salida es un objeto Text que representa el evento, y el valor de salida es un objeto Text que representa el nombre del atleta.

El método map divide cada línea de datos de entrada por el delimitador de coma, extrae el nombre del atleta y el evento, y emite un par clave-valor con el evento como clave y el nombre del atleta como valor.

Implementar el Particionador

En este paso, crearemos una clase Particionador personalizada que particione los pares clave-valor en función del evento.

Primero, cree un archivo Java para la clase Particionador:

touch /home/hadoop/OlympicPartitioner.java

Luego, agregue el siguiente código al archivo 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);
    }
}

La clase OlympicPartitioner extiende la clase Partitioner proporcionada por Hadoop. Sobrescribe el método getPartition, que toma la clave, el valor y el número de particiones como entrada.

El método getPartition calcula un código hash para el evento (clave) y devuelve el número de partición tomando el valor absoluto del código hash módulo el número de particiones. Esto garantiza que todos los registros con el mismo evento se envíen a la misma partición para su procesamiento por el Reducer.

Implementar el Reductor

En este paso, crearemos una clase Reductor que procese los datos particionados y genere la salida final.

Primero, cree un archivo Java para la clase Reductor:

touch /home/hadoop/OlympicReducer.java

Luego, agregue el siguiente código al archivo 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()));
    }
}

La clase OlympicReducer extiende la clase Reducer proporcionada por Hadoop. Define la clave de entrada como Text (que representa el evento), el valor de entrada como Text (que representa el nombre del atleta) y la clave y el valor de salida como objetos Text.

El método reduce se llama para cada clave de evento única, con un iterador sobre los nombres de los atletas asociados con ese evento. Construye una lista separada por comas de los atletas para cada evento y emite un par clave-valor con el evento como clave y la lista de atletas como valor.

Escribir el Controlador

En este paso, crearemos una clase Controlador que enlaza las clases Mapper, Particionador y Reductor y ejecuta el trabajo MapReduce.

Primero, cree un archivo Java para la clase Controlador:

touch /home/hadoop/OlympicDriver.java

Luego, agregue el siguiente código al archivo 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);
    }
}

La clase OlympicDriver es el punto de entrada del trabajo MapReduce. Configura la configuración del trabajo, especifica las clases Mapper, Particionador y Reductor y configura las rutas de entrada y salida.

En el método main, creamos un nuevo objeto Configuration y una instancia de Job con el nombre del trabajo "Olympic Partitioner". Establecemos las clases Mapper, Particionador y Reductor utilizando los métodos setter correspondientes.

También establecemos las clases de clave y valor de salida como Text. Las rutas de entrada y salida se especifican utilizando los argumentos de línea de comandos pasados al controlador.

Finalmente, llamamos al método waitForCompletion en la instancia de Job para ejecutar el trabajo MapReduce y salir con un código de estado adecuado (0 para éxito, 1 para error).

Para ejecutar el trabajo, debe compilar las clases Java y crear un archivo jar. Luego, puede ejecutar el archivo jar utilizando el siguiente 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 los resultados ejecutando el siguiente comando:

hadoop fs -cat /output/*

Salida de ejemplo:

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

Resumen

En este laboratorio, exploramos el concepto del Particionador de Hadoop Shuffle al diseñar un escenario inspirado en la antigua Olimpiada griega. Implementamos una clase Mapper para leer los datos de entrada y generar pares clave-valor, una clase Particionador personalizada para particionar los datos en función del evento y una clase Reductor para procesar los datos particionados y generar la salida final.

A través de este laboratorio, adquirí experiencia práctica con el modelo de programación MapReduce y aprendí cómo aprovechar la clase Particionador para distribuir eficientemente los datos entre las particiones. El escenario de la antigua Olimpiada griega proporcionó un contexto atractivo para comprender las aplicaciones prácticas del Particionador de Shuffle en un entorno real.