项目4 MapReduce 多维度数据分析完整教程

项目概述

本教程将带你深入学习 MapReduce 编程框架,通过一个实际的电影网站用户影评分析项目,掌握从基础到进阶的 MapReduce 开发技能。

项目背景

随着电影市场的日益扩大和竞争加剧,如何从海量用户评论中快速提取有价值的信息,成为电影制作方、发行方以及在线播放平台共同面临的挑战。本项目将使用 MapReduce 实现多维度分析电影网站用户影评。

学习目标

• 掌握 MapReduce 输入输出格式设置
• 学会使用 Hadoop Java API 进行文件操作
• 理解并应用 MapReduce 优化技术
• 在 IntelliJ IDEA 中高效开发和部署 MapReduce 程序

---

第一部分:MapReduce 输入输出格式

1.1 为什么需要序列化格式?

在大数据处理中,尤其是处理逻辑复杂的情况下,需要使用多个 MapReduce 程序连续处理,这会在 HDFS 中保存大量中间结果。提高中间结果的存取效率,对整个数据处理流程具有重要意义。

Hadoop 序列化的特点:

• 紧凑: 高效使用存储空间
• 快速: 读取数据的额外开销小
• 可扩展: 可透明地读取旧格式的数据
• 多语言交互: 可以使用不同语言读/写永久存储的数据

1.2 常用输入格式

输入格式	描述	键类型	值类型
TextInputFormat	默认格式,读取文件的行	LongWritable(行的字节偏移量)	Text(行的内容)
SequenceFileInputFormat	Hadoop 定义的高性能二进制格式	用户自定义	用户自定义
KeyValueTextInputFormat	将行解析为键值对	Text(第一个制表符前的字符)	Text(行剩下的内容)

设置输入格式:

job.setInputFormatClass(SequenceFileInputFormat.class);

1.3 常用输出格式

输出格式	描述
TextOutputFormat	默认格式,以键值对形式输出行
SequenceFileOutputFormat	输出二进制文件(序列化文件),适合作为后续 MapReduce 任务的输入
NullOutputFormat	忽略收到的数据,即没有输出

设置输出格式:

job.setOutputFormatClass(SequenceFileOutputFormat.class);

1.4 实战任务:筛选日志并生成序列化文件

任务目标: 筛选出竞赛网站 2024 年 1 月和 2 月的用户访问日志,以序列化格式输出。

关键代码:

// 设置输入输出格式
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(SequenceFileOutputFormat.class);

// Mapper 中筛选数据
if (date.startsWith("2024-01") || date.startsWith("2024-02")) {
    context.write(new Text(userID), new Text(date));
}

---

第二部分:Hadoop Java API 文件操作

2.1 获取 FileSystem 实例

FileSystem 是 Hadoop 通用的文件系统 API,提供三种获取实例的方法:

// 方法1: 返回默认文件系统
FileSystem fs = FileSystem.get(conf);

// 方法2: 通过 URI 指定文件系统
FileSystem fs = FileSystem.get(uri, conf);

// 方法3: 指定用户,更安全
FileSystem fs = FileSystem.get(uri, conf, "username");

2.2 目录管理操作

查看目录内容:

FileStatus[] fileStatuses = fs.listStatus(new Path("/user/root"));
for (FileStatus status : fileStatuses) {
    if (status.isDirectory()) {
        System.out.println("目录: " + status.getPath().getName());
    } else {
        System.out.println("文件: " + status.getPath().getName());
    }
}

创建目录:

// 创建目录及其父目录
boolean success = fs.mkdirs(new Path("/user/root/newdir"));

// 创建目录并设置权限
FsPermission permission = new FsPermission("755");
fs.mkdirs(new Path("/user/root/newdir"), permission);

2.3 文件操作

删除文件:

// 删除文件或空目录
fs.delete(new Path("/user/root/file.txt"), false);

// 递归删除目录
fs.delete(new Path("/user/root/dir"), true);

上传文件:

// 从本地上传到 HDFS
fs.copyFromLocalFile(
    new Path("/local/path/file.txt"),
    new Path("/hdfs/path/file.txt")
);

// 上传后删除本地文件
fs.copyFromLocalFile(
    true,  // delSrc
    new Path("/local/path/file.txt"),
    new Path("/hdfs/path/file.txt")
);

下载文件:

// 从 HDFS 下载到本地
fs.copyToLocalFile(
    new Path("/hdfs/path/file.txt"),
    new Path("/local/path/file.txt")
);

2.4 读写数据流

读取数据:

FSDataInputStream in = fs.open(new Path("/user/root/file.txt"));
BufferedReader reader = new BufferedReader(new InputStreamReader(in));
String line;
while ((line = reader.readLine()) != null) {
    System.out.println(line);
}
reader.close();

写入数据:

FSDataOutputStream out = fs.create(new Path("/user/root/output.txt"));
BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));
writer.write("Hello Hadoop!");
writer.close();

2.5 读取序列化文件

使用 SequenceFile.Reader:

SequenceFile.Reader reader = new SequenceFile.Reader(conf,
    SequenceFile.Reader.file(new Path("/path/to/seqfile")));

// 获取键值类型
System.out.println("Key Class: " + reader.getKeyClassName());
System.out.println("Value Class: " + reader.getValueClassName());

// 读取数据
Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf);

while (reader.next(key, value)) {
    System.out.println(key + ": " + value);
}
reader.close();

---

第三部分:优化 MapReduce 程序

3.1 自定义键值类型

Hadoop 提供了丰富的内置数据类型,但有时需要自定义类型来满足特定需求。

自定义值类型(实现 Writable 接口):

public class CustomValue implements Writable {
    private String field1;
    private int field2;
    
    // 序列化:写入数据
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(field1);
        out.writeInt(field2);
    }
    
    // 反序列化:读取数据
    @Override
    public void readFields(DataInput in) throws IOException {
        field1 = in.readUTF();
        field2 = in.readInt();
    }
    
    // 重写 toString 方法,方便输出
    @Override
    public String toString() {
        return field1 + "\t" + field2;
    }
}

自定义键类型(实现 WritableComparable 接口):

public class CustomKey implements WritableComparable<CustomKey> {
    private String userID;
    private String date;
    
    // 序列化
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(userID);
        out.writeUTF(date);
    }
    
    // 反序列化
    @Override
    public void readFields(DataInput in) throws IOException {
        userID = in.readUTF();
        date = in.readUTF();
    }
    
    // 比较方法,用于排序
    @Override
    public int compareTo(CustomKey o) {
        int result = this.userID.compareTo(o.userID);
        if (result == 0) {
            return this.date.compareTo(o.date);
        }
        return result;
    }
}

3.2 Combiner 优化

什么是 Combiner? Combiner 是在 Map 端进行局部聚合的组件,可以大幅减少网络传输的数据量。

使用场景:

• ✅ 求和、求最大值、求最小值
• ❌ 求平均值(会导致结果错误)

原因分析:

求和: sum(1,1,1,1) = sum(sum(1,1), sum(1,1)) = sum(2,2) = 4 ✓

求平均值: mean(4,5,3,6,2) ≠ mean(mean(4,5), mean(3,6,2)) 
         = mean(4.5, 5.5) = 5 ✗

实现 Combiner:

public class MyCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable value : values) {
            sum += value.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

// 在驱动类中配置
job.setCombinerClass(MyCombiner.class);

3.3 Partitioner 分区

什么是 Partitioner? Partitioner 根据业务需求将 Map 输出的键值对分发到不同的 Reducer。

默认分区器 HashPartitioner:

public int getPartition(K2 key, V2 value, int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
}

自定义 Partitioner:

public class MonthPartitioner extends Partitioner<Text, IntWritable> {
    @Override
    public int getPartition(Text key, IntWritable value, int numReduceTasks) {
        String date = key.toString();
        
        // 根据月份分区
        if (date.contains("2024-01")) {
            return 0;
        } else if (date.contains("2024-02")) {
            return 1;
        }
        return 0;
    }
}

// 在驱动类中配置
job.setPartitionerClass(MonthPartitioner.class);
job.setNumReduceTasks(2);  // 设置 Reducer 数量

3.4 自定义计数器

使用枚举定义计数器:

// 定义计数器枚举
enum MyCounters {
    VALID_RECORDS,
    INVALID_RECORDS
}

// 在 Mapper 中使用
context.getCounter(MyCounters.VALID_RECORDS).increment(1);

使用动态计数器:

// 根据字段值动态创建计数器
context.getCounter("CustomGroup", fieldValue).increment(1);

查看计数器输出:

Job Counters
    CustomGroup
        VALID_RECORDS=1000
        INVALID_RECORDS=50

---

第四部分:在 IntelliJ IDEA 中开发部署

4.1 传递参数

在驱动类中设置参数:

Configuration conf = new Configuration();
conf.set("inputPath", "/user/root/input");
conf.set("outputPath", "/user/root/output");

在 Mapper/Reducer 中获取参数:

public class MyMapper extends Mapper<...> {
    private String inputPath;
    
    @Override
    protected void setup(Context context) {
        Configuration conf = context.getConfiguration();
        inputPath = conf.get("inputPath");
    }
}

4.2 使用 ToolRunner

实现 Tool 接口:

public class MyDriver extends Configured implements Tool {
    
    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        Job job = Job.getInstance(conf, "My Job");
        
        // 配置 MapReduce 作业
        job.setJarByClass(MyDriver.class);
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);
        
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        return job.waitForCompletion(true) ? 0 : 1;
    }
    
    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new Configuration(), new MyDriver(), args);
        System.exit(exitCode);
    }
}

4.3 远程连接 Hadoop 集群

配置集群连接:

public static Configuration getMyConfiguration() {
    Configuration conf = new Configuration();
    String resourceNode = "master";
    
    // 指定 NameNode
    conf.set("fs.defaultFS", "hdfs://master:8020");
    
    // 指定使用 YARN 框架
    conf.set("mapreduce.framework.name", "yarn");
    
    // 指定 ResourceManager
    conf.set("yarn.resourcemanager.address", resourceNode + ":8032");
    conf.set("yarn.resourcemanager.scheduler.address", resourceNode + ":8030");
    
    // 指定 HistoryServer
    conf.set("mapreduce.jobhistory.address", resourceNode + ":10020");
    
    // 支持跨平台提交
    conf.setBoolean("mapreduce.app-submission.cross-platform", true);
    
    // 指定 JAR 包路径
    conf.set("mapreduce.job.jar", "target/MyJob.jar");
    
    return conf;
}

在驱动类中使用:

@Override
public int run(String[] args) throws Exception {
    Configuration conf = getMyConfiguration();  // 获取集群配置
    Job job = Job.getInstance(conf, "My Job");
    // ... 其他配置
}

4.4 自动打包程序

创建打包工具类:

public class JarUtil {
    public static String jar(Class<?> clazz) throws Exception {
        String localJarPath = "target/" + clazz.getSimpleName() + ".jar";
        
        // 创建 JAR 文件
        FileOutputStream fos = new FileOutputStream(localJarPath);
        JarOutputStream jos = new JarOutputStream(fos);
        
        // 添加类文件到 JAR
        String classPath = clazz.getName().replace(".", "/") + ".class";
        jos.putNextEntry(new JarEntry(classPath));
        
        // 写入类文件内容
        InputStream is = clazz.getResourceAsStream("/" + classPath);
        byte[] buffer = new byte[1024];
        int len;
        while ((len = is.read(buffer)) > 0) {
            jos.write(buffer, 0, len);
        }
        
        is.close();
        jos.closeEntry();
        jos.close();
        
        return localJarPath;
    }
}

在配置中使用:

conf.set("mapreduce.job.jar", JarUtil.jar(MyDriver.class));

注意事项:

• 需要将 Hadoop 配置文件 yarn-site.xml 复制到项目的 resources 目录
• 确保网络可以连接到 Hadoop 集群
• 在 IDEA 中直接运行 main 方法即可提交任务

---

第五部分:项目实战 - 电影影评分析

5.1 任务1: 统计评分次数最多的10部电影

分析思路:

1. 使用 Map Join 连接电影信息和评分信息
2. 统计每部电影的评分次数
3. 按评分次数降序排序
4. 取前10部电影

关键代码:

// Mapper: 统计评分次数
public class RateCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] fields = value.toString().split("::");
        String movieID = fields[1];  // 电影ID
        context.write(new Text(movieID), new IntWritable(1));
    }
}

// Reducer: 汇总评分次数
public class RateCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
        int count = 0;
        for (IntWritable value : values) {
            count += value.get();
        }
        context.write(key, new IntWritable(count));
    }
}

// 第二个 MR: 排序并取 Top 10
public class Top10Mapper extends Mapper<LongWritable, Text, IntWritable, Text> {
    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] fields = value.toString().split("\t");
        String movieID = fields[0];
        int count = Integer.parseInt(fields[1]);
        // 使用负数实现降序排序
        context.write(new IntWritable(-count), new Text(movieID));
    }
}

public class Top10Reducer extends Reducer<IntWritable, Text, Text, IntWritable> {
    private int count = 0;
    
    @Override
    protected void reduce(IntWritable key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        for (Text value : values) {
            if (count < 10) {
                context.write(value, new IntWritable(-key.get()));
                count++;
            }
        }
    }
}

5.2 任务2: 不同性别用户评分最高的10部电影

分析思路:

1. Join 用户信息、评分信息和电影信息
2. 按性别和电影分组,计算平均评分
3. 在每个性别组内按评分降序排序
4. 每组取前10部电影

自定义分组和排序:

// 自定义键类型
public class GenderMovieKey implements WritableComparable<GenderMovieKey> {
    private String gender;
    private double avgRate;
    
    // 排序规则: 先按性别升序,再按评分降序
    @Override
    public int compareTo(GenderMovieKey o) {
        int result = this.gender.compareTo(o.gender);
        if (result == 0) {
            return Double.compare(o.avgRate, this.avgRate);  // 降序
        }
        return result;
    }
    
    // 省略 write 和 readFields 方法
}

// 自定义分组器
public class GenderGroupingComparator extends WritableComparator {
    protected GenderGroupingComparator() {
        super(GenderMovieKey.class, true);
    }
    
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        GenderMovieKey k1 = (GenderMovieKey) a;
        GenderMovieKey k2 = (GenderMovieKey) b;
        // 只按性别分组
        return k1.getGender().compareTo(k2.getGender());
    }
}

// Reducer: 每组取前10
public class Top10ByGenderReducer extends Reducer<GenderMovieKey, Text, Text, DoubleWritable> {
    private int count = 0;
    private String currentGender = "";
    
    @Override
    protected void reduce(GenderMovieKey key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        // 检测性别变化,重置计数器
        if (!key.getGender().equals(currentGender)) {
            currentGender = key.getGender();
            count = 0;
        }
        
        for (Text value : values) {
            if (count < 10) {
                context.write(value, new DoubleWritable(key.getAvgRate()));
                count++;
            }
        }
    }
}

// 驱动类配置
job.setGroupingComparatorClass(GenderGroupingComparator.class);

5.3 任务3: 指定电影各年龄段用户的平均评分

数据准备:

• Age 字段取值: 0(18岁以下), 1(18-24), 2(25-34), 3(35-44), 4(45-49), 5(50-55), 6(56+)

实现代码:

// Mapper: 筛选指定电影并输出年龄段和评分
public class AgeAvgMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private String targetMovieID = "2858";  // 评分次数最多的电影
    
    @Override
    protected void setup(Context context) {
        Configuration conf = context.getConfiguration();
        targetMovieID = conf.get("movieID", "2858");
    }
    
    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] fields = value.toString().split("::");
        String movieID = fields[1];
        
        if (movieID.equals(targetMovieID)) {
            String age = fields[4];      // 年龄段
            int rating = Integer.parseInt(fields[2]);  // 评分
            context.write(new Text(age), new IntWritable(rating));
        }
    }
}

// Reducer: 计算平均评分
public class AgeAvgReducer extends Reducer<Text, IntWritable, Text, DoubleWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
        int sum = 0;
        int count = 0;
        
        for (IntWritable value : values) {
            sum += value.get();
            count++;
        }
        
        double avg = (double) sum / count;
        context.write(key, new DoubleWritable(avg));
    }
}

5.4 任务4: 各电影类型中评分最高的5部电影

实现思路:

1. 先按电影类型和ID分组,计算每部电影的平均评分
2. 在每个类型组内按评分降序排序
3. 每组取前5部电影

关键代码:

// 第一步: 计算每部电影的平均评分
public class TypeAvgMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String[] fields = value.toString().split("::");
        String movieType = fields[3];  // 电影类型
        String movieID = fields[0];
        int rating = Integer.parseInt(fields[2]);
        
        // 以"类型::电影ID"作为键
        context.write(new Text(movieType + "::" + movieID), new IntWritable(rating));
    }
}

// 第二步: 自定义键类型实现二次排序
public class TypeMovieKey implements WritableComparable<TypeMovieKey> {
    private String movieType;
    private double avgRate;
    
    @Override
    public int compareTo(TypeMovieKey o) {
        int result = this.movieType.compareTo(o.movieType);
        if (result == 0) {
            // 同类型内按评分降序
            return Double.compare(o.avgRate, this.avgRate);
        }
        return result;
    }
}

// 自定义分组器
public class TypeGroupingComparator extends WritableComparator {
    protected TypeGroupingComparator() {
        super(TypeMovieKey.class, true);
    }
    
    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        TypeMovieKey k1 = (TypeMovieKey) a;
        TypeMovieKey k2 = (TypeMovieKey) b;
        // 只按电影类型分组
        return k1.getMovieType().compareTo(k2.getMovieType());
    }
}

// Reducer: 每组取前5
public class Top5ByTypeReducer extends Reducer<TypeMovieKey, Text, Text, DoubleWritable> {
    private int count = 0;
    private String currentType = "";
    
    @Override
    protected void reduce(TypeMovieKey key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        if (!key.getMovieType().equals(currentType)) {
            currentType = key.getMovieType();
            count = 0;
        }
        
        for (Text value : values) {
            if (count < 5) {
                context.write(new Text(currentType + "::" + value.toString()), 
                            new DoubleWritable(key.getAvgRate()));
                count++;
            }
        }
    }
}

---

第六部分:最佳实践与优化建议

6.1 性能优化清单

1. 合理使用 Combiner

• ✅ 适用场景: 求和、计数、最大/最小值
• ❌ 不适用: 求平均值、中位数等需要全局计算的场景

2. 选择合适的数据格式

• 中间结果使用 SequenceFile 格式
• 最终输出根据需求选择 Text 或 SequenceFile

3. 自定义 Partitioner

• 避免数据倾斜
• 让数据均匀分布到各个 Reducer

4. 压缩中间结果

conf.setBoolean("mapreduce.map.output.compress", true);
conf.setClass("mapreduce.map.output.compress.codec", 
              SnappyCodec.class, CompressionCodec.class);

5. 调整 JVM 参数

conf.set("mapreduce.map.java.opts", "-Xmx2048m");
conf.set("mapreduce.reduce.java.opts", "-Xmx4096m");

6.2 开发调试技巧

1. 使用本地模式调试

conf.set("mapreduce.framework.name", "local");
conf.set("fs.defaultFS", "file:///");

2. 添加日志输出

System.err.println("Debug info: " + value);  // 输出到任务日志

3. 使用计数器监控

context.getCounter("MyGroup", "ProcessedRecords").increment(1);

4. 小数据集测试

• 先用小数据集验证逻辑
• 确认无误后再处理大数据集

6.3 常见问题解决

问题1: 任务卡在 Map 0%

• 检查输入路径是否正确
• 检查 NameNode 和 ResourceManager 是否正常

问题2: OutOfMemoryError

• 增加 JVM 堆内存
• 减小输入分片大小
• 优化代码,避免内存泄漏

问题3: 数据倾斜

• 自定义 Partitioner
• 增加 Reducer 数量
• 对热点键进行预聚合

问题4: 远程提交失败

• 检查集群配置是否正确
• 确认 JAR 包路径设置
• 检查网络连接

---

总结

本教程系统地介绍了 MapReduce 编程的核