介绍

数据如下：

详细说明、下载见链接

过滤要求

根据 tab 制表符拆分每行数据，结果是数组
如果数组有 6 个元素，且第 3 个元素中包含数字，就符合过滤要求
将符合过滤要求的第 2 个元素和第 3 个元素记录到结果文件中

工程

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

java代码

mapper类

package top.malaoshi.filter;

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

/**
 * 继承自 Mapper 类，用于过滤和处理包含特定关键字的数据。
 * 泛型解释：
 * LongWritable：每一行开始位置的偏移量，对应 map() 方法第1个形参类型
 * Text：输入数据的值，通常是一行文本数据，对应 map() 方法第2个形参类型
 * Text：context写入的key的数据类型
 * Text：context写入的value的数据类型
 */
public class FilterMapper extends Mapper<LongWritable, Text, Text, Text> {
    // hadoop 实现序列化好处之二：复用同一个对象
    private Text k = new Text();
    private Text v = new Text();
    /**
     * map 方法是MapReduce的核心部分之一，它接收输入的数据并进行处理。
     *
     * @param key 是每一行开始位置的偏移量
     * @param value 输入数据的值，通常是一行文本数据。
     * @param context 上下文对象，用于写入中间键值对。
     */
    protected void map(LongWritable  key, Text value, Context context) throws IOException, InterruptedException {
        // 将输入的值转换为字符串，并以制表符分割成数组
        String[] str = value.toString().split("\t");

        // 检查分割后的数组长度是否为6
        if (str != null && str.length == 6) {
            // 提取关键词和用户ID
            String uid = str[1];      // 用户ID位于第二个位置
            String keyword = str[2];  // 关键词位于第三个位置

            // 使用正则表达式检查关键词是否包含数字
//            开头的 .*：匹配数字出现前的「任意字符（0 个或多个）」（比如空、字母、符号等）；
//            中间的 \d+：匹配「至少 1 个连续数字」（这是核心，必须存在）；
//            结尾的 .*：匹配数字出现后的「任意字符（0 个或多个）」；
            if (keyword.matches(".*\\d+.*")) {  // 如果关键词包含至少一个数字
                // 输出<uid, keyword>作为中间键值对
                k.set(uid);
                v.set(keyword);
                context.write(k, v);
            }
        }
    }
}

reduce类

只需要通过 mapper 实现过滤，没有统计的需求，所以不需要reduce类

主启动类-driver

package top.malaoshi.filter;

import java.io.IOException;  // 引入异常处理相关的类

// 引入Hadoop相关配置及API包
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 Main {
    public static void main(String[] args) throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
        // 检查命令行参数，确保提供了输入和输出路径
        if (args != null && args.length == 2) {
            // 创建Hadoop Configuration对象，用于加载配置信息
            Configuration conf = new Configuration();

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

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

            // 设置Map阶段使用的Mapper类
            job.setMapperClass(FilterMapper.class);

            // 设置输出key-value对的数据类型（此处均为Text）
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.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);
        } else {
            // 如果没有提供正确的参数，则打印错误信息
            System.err.println("<Usage> wordcount <input> <output>");
        }
    }
}