介绍

数据如下：

详细说明、下载见链接

过滤要求

根据 tab 制表符拆分每行数据，结果是数组
如果数组有 6 个元素，就将第 2 列 UID 去重，并写到结果文件中

工程

在 hadoop3.x mapreduce 实现wordcount（本机运行）工程进行操作

java代码

mapper类

package top.malaoshi.uid;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
/**
 * 继承自 Mapper 类，用于过滤和处理包含特定关键字的数据。
 * 泛型解释：
 * LongWritable：每一行开始位置的偏移量，对应 map() 方法第1个形参类型
 * Text：输入数据的值，通常是一行文本数据，对应 map() 方法第2个形参类型
 * Text：context写入的key的数据类型
 * NullWritable：相当于null，context写入的value的数据类型
 */
public class UidMapper extends Mapper<LongWritable,Text,Text,NullWritable> {
    // hadoop 实现序列化好处之二：复用同一个对象
    private Text k = new Text();
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 必须通过下面方式转成字符串，否则可能中文乱码
        String newValue = new String(value.getBytes(),"utf-8");
        // 以制表符分割成数组
        String[] str = newValue.split("\t");

        // 检查分割后的数组长度是否为6
        if (str != null && str.length == 6) {
            // 提取关键词和用户ID
            String uid = str[1];      // 用户ID位于第二个位置
            k.set(uid);
            // 由于是对 uid 去重，所以不需要value，写入 NullWritable，不从数据流中读数据，也不写入数据
            context.write(k, NullWritable.get());
        }

    }
}

reduce类

package top.malaoshi.uid;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * 继承自 Reducer 类
 * 泛型解释：
 * Text：Mapper 输出的 key 类型，也对应 reduce() 方法第1个形参类型
 * NullWritable：Mapper 输出的 value 类型，对应 reduce() 方法第2个形参类型
 * Text：context写入的 key 的数据类型
 * NullWritable：context写入的 value 的数据类型。相当于null，不会向外输出数据
 */
public class UidReducer extends Reducer<Text,NullWritable,Text,NullWritable> {
    @Override
    protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        // 由于是 uid 去重，最后只需要将uid写到文件中。
        // 不需要value，所以value是 NullWritable，不从数据流中读数据，也不写入数据
        context.write(key, NullWritable.get());
    }
}

主启动类-driver

package top.malaoshi.uid;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class Main {
    public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
        // 检查命令行参数，确保提供了输入和输出路径
        if (args.length != 2 || args == null){ // 检查命令行参数是否正确（需要两个参数）
            System.err.println("please input current Path"); // 提示用户输入路径
            System.exit(1); // 参数不正确时退出程序
        }
        // 创建Hadoop Configuration对象，用于加载配置信息
        Configuration conf = new Configuration();

        // 创建一个Job实例，指定任务名
        Job job = new Job(conf, "UID去重");

        // 设置主类，以便在执行时能够找到该类。这允许Hadoop以jar包形式运行此程序
        job.setJarByClass(Main.class);

        // 设置Map阶段使用的Mapper类
        job.setMapperClass(UidMapper.class);
        // 设置Reduce阶段使用的reduce类
        job.setReducerClass(UidReducer.class);

        // 设置 Mapper 输出key-value对的数据类型，必须与 Mapper 泛型一致
        // 与 job.setOutputKeyClass()、job.setOutputValueClass() 一致时可略
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        // 设置输出key-value对的数据类型，必须与 Reducer 泛型一致
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        // 添加输入路径到job中
        FileInputFormat.addInputPath(job, new Path(args[0]));

        // 设置输出路径到job中，注意：该目录不应存在
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        // 提交job，并等待完成。根据job是否成功完成返回状态码
        System.exit(job.waitForCompletion(true) ? 0 : 1);

    }
}