介绍

数据如下：

详细说明、下载见链接

过滤要求

统计的是每个用户，搜索 电影 的记录：

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

注意：数据文件第 2 列是用户UID，同一个用户，每搜索一次 电影，就会存在一行记录，所以会存在重复的 UID

工程

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

java代码

mapper类

package top.malaoshi.filter2;

import org.apache.hadoop.io.LongWritable;
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的数据类型
 * 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 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位于第二个位置
            String keyword = str[2];  // 关键词位于第三个位置

            // 使用正则表达式检查关键词是否包含数字
            if (keyword.contains("电影")) {  // 如果关键词包含电影
                // 输出<uid, keyword>作为中间键值对
                k.set(uid);
                v.set(keyword);
                context.write(k, v);
            }
        }

    }
}

reduce类

package top.malaoshi.filter2;

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

import java.io.IOException;

/**
 * 继承自 Reducer 类
 * 泛型解释：
 * Text：Mapper 输出的key类型，也对应 reduce() 方法第1个形参类型
 * Text：Mapper 输出的 value 类型，对应 reduce() 方法第2个形参类型
 * Text：context写入的key的数据类型
 * Text：context写入的value的数据类型
 */
public class FilterReduce extends Reducer<Text,Text,Text,Text> {
    Text v = new Text();
    @Override
    protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {

        StringBuilder stringBuilder = new StringBuilder();
        for(Text item:values){
//            注意：这里用下面方式转成字符串，否则数据错误
            String newValue = new String(item.getBytes(),0,item.getLength(),"utf-8");
            stringBuilder.append(newValue);
            stringBuilder.append(",");
        }
        int len = stringBuilder.length();
        stringBuilder.delete(len - 1, len );
        v.set(stringBuilder.toString());
        context.write(key,v);
    }
}

主启动类-driver

package top.malaoshi.filter2;

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;

import java.io.IOException;

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, "统计每个用户包含指定关键字的记录");

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

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

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

            // 设置输出key-value对的数据类型，必须与 Reducer 泛型一致
            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>");
        }
    }
}