La eliminación de duplicados es una operación fundamental en el procesamiento distribuido de grandes volúmenes de datos. Su propósito principal consiste en identificar y conservar únicamente una instencia de cada registro único, lo que resulta crítico en escenarios como la deduplicación de identificadores de productos, la normalización de claves de usuario o la extracción de valores distintos a partir de registros de logs.

Desde una perspectiva arquitectónica de MapReduce, esta tarea se resuelve aprovechando la fase de shuffle and sort: al emitir el mismo valor como clave desde múltiples mappers, el framework agrupa automáticamente todos los valores asociados bajo esa clave antes de invocar el reducer. Por tanto, basta con usar el campo objetivo (por ejemplo, el ID del producto) como clave en la salida del mapper, ignorando los valores —ya que no se requiere agregación— y simplemente emitir la clave una sola vez en el reducer.

Entorno de ejecución:

• Sistema operativo: CentOS 7
• JDK: versión 1.8
• Hadoop: 3.2.0 (en modo clúster)
• IDE: IntelliJ IDEA 2019

Datos de entrada:

Un conjunto tabular con tres columnas: user_id, product_id y favorite_timestamp, separadas por tabuladores. A continuación, un extracto representativo:

10181	1000481	2010-04-04 16:54:31
20001	1001597	2010-04-07 15:07:52
20001	1001560	2010-04-07 15:08:27
20042	1001368	2010-04-08 08:20:30
20067	1002061	2010-04-08 16:45:33
20056	1003289	2010-04-12 10:50:55
...

Objetivo: generar una lista única de product_id, eliminando todas las repeticiones.

Preparación previa:

1. Iniciar el clúster Hadoop:
   ./start-all.sh
2. Crear directorios en HDFS y cargar los datos:
   hadoop fs -mkdir -p /dedupe/input
hadoop fs -put buyer_favorite1 /dedupe/input/

Implementación Java optimizada (con nombres descriptivos, estructura limpia y sin efectos secundarios innecesarios):

package com.example.dedupe;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;

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

public class DistinctProductExtractor {

    public static class TokenizerMapper extends Mapper<Object, Text, Text, NullWritable> {
        private final Text productId = new Text();

        @Override
        protected void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString().trim();
            if (line.isEmpty()) return;

            String[] fields = line.split("\t", -1);
            if (fields.length > 1 && !fields[1].isEmpty()) {
                productId.set(fields[1]);
                context.write(productId, NullWritable.get());
            }
        }
    }

    public static class DeduplicateReducer extends Reducer<Text, NullWritable, Text, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context)
                throws IOException, InterruptedException {
            context.write(key, NullWritable.get());
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "distinct-product-dedupe");
        job.setJarByClass(DistinctProductExtractor.class);

        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(DeduplicateReducer.class); // Optimización opcional
        job.setReducerClass(DeduplicateReducer.class);

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

        FileInputFormat.addInputPath(job, new Path("hdfs://localhost:9000/dedupe/input/buyer_favorite1"));
        FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost:9000/dedupe/output"));

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

Verificación del resultado:

hadoop fs -ls /dedupe/output
hadoop fs -cat /dedupe/output/part-r-00000

La salida conetndrá una lista ordanada lexicográficamente (por el comportamiento implícito de shuffle) de todos los product_id únicos presentes en el conjunto original.