手把手教你用Hadoop MapReduce搞定手机流量统计（附完整Java代码）

从零开始实战：用Hadoop MapReduce实现手机流量统计

最近在整理旧手机账单时，突然好奇自己一年到底用了多少流量。作为程序员，第一反应不是去营业厅查记录，而是想着能不能写个程序统计出来。正好手头有Hadoop环境，不如用MapReduce来实现这个需求。本文将带你从环境搭建到代码调试，完整实现手机流量统计功能。

1. 环境准备与数据理解

在开始编码前，我们需要确保开发环境就绪，并充分理解待处理的数据结构。这是很多初学者容易忽略的关键步骤。

1.1 Hadoop环境配置

对于本地开发和测试，推荐以下两种环境配置方案：

方案一：本地模式（无需HDFS）

• 下载Hadoop二进制包（3.x版本）
• 解压后设置环境变量：

  export HADOOP_HOME=/path/to/hadoop export PATH=$PATH:$HADOOP_HOME/bin

• 验证安装：

  hadoop version

方案二：伪分布式模式（单节点HDFS）

• 修改etc/hadoop/core-site.xml：

  <configuration> <property> <name>fs.defaultFS</name> <value>hdfs://localhost:9000</value> </property> </configuration>

• 格式化HDFS并启动服务：

  hdfs namenode -format start-dfs.sh

提示：Windows用户建议使用WSL2或虚拟机运行Hadoop，避免原生Windows环境下的兼容性问题。

1.2 数据结构分析

我们的原始数据phonetraffic.txt格式如下：

18632845069,Jan,40978,94715 18632845069,Feb,39481,63612 ...

每行包含4个字段，用逗号分隔：

1. 手机号码（如18632845069）
2. 月份缩写（如Jan）
3. 上行流量（单位KB）
4. 下行流量（单位KB）

数据特点：

• 每月一条记录，全年共12个月
• 流量值为整数，无小数
• 手机号作为唯一标识符
• 总流量=上行+下行

2. MapReduce程序设计原理

理解MapReduce的工作原理比直接写代码更重要。让我们先拆解这个统计任务的逻辑流程。

2.1 计算模型分解

对于手机流量统计，MapReduce的处理流程可分为三个阶段：

1. Map阶段：

   • 输入：原始数据行（文本格式）
   • 处理：解析每行数据，计算单月总流量
   • 输出：键值对<手机号, 当月总流量>

2. Shuffle阶段（自动完成）：

   • 将相同手机号的数据发送到同一个Reducer
   • 排序并分组键值对

3. Reduce阶段：

   • 输入：<手机号, [当月流量1, 当月流量2...]>
   • 处理：累加所有月份流量
   • 输出：<手机号, 年度总流量>

2.2 关键类与数据类型

Hadoop使用特定的可序列化类型替代Java原生类型：

在流量统计中：

• Map输出键：Text（手机号）
• Map输出值：IntWritable（单月流量）
• Reduce输出键：Text（手机号）
• Reduce输出值：IntWritable（年度流量）

3. 完整代码实现与逐行解析

现在我们来编写完整的MapReduce程序，我会详细解释每个关键部分的实现逻辑。

3.1 项目结构与依赖

创建Maven项目，添加Hadoop依赖：

<dependencies> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.3.4</version> </dependency> </dependencies>

3.2 Mapper实现

public static class TrafficMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private Text phoneNumber = new Text(); private IntWritable monthlyTraffic = new IntWritable(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 1. 分割CSV行 String[] fields = value.toString().split(","); if (fields.length != 4) return; // 跳过格式错误的行 // 2. 提取手机号 String number = fields[0].trim(); // 3. 计算单月总流量（上行+下行） try { int upload = Integer.parseInt(fields[2].trim()); int download = Integer.parseInt(fields[3].trim()); int total = upload + download; // 4. 输出键值对 phoneNumber.set(number); monthlyTraffic.set(total); context.write(phoneNumber, monthlyTraffic); } catch (NumberFormatException e) { // 忽略数值解析错误 } } }

关键点解析：

• LongWritable key：输入的行偏移量，通常不直接使用
• 防御性编程：处理可能的格式错误和数值异常
• context.write()：发射键值对到Reduce阶段

3.3 Reducer实现

public static class TrafficReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable yearlyTraffic = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { // 1. 初始化年度总量 int sum = 0; // 2. 遍历所有月份数据 for (IntWritable value : values) { sum += value.get(); } // 3. 输出结果 yearlyTraffic.set(sum); context.write(key, yearlyTraffic); } }

优化技巧：

• 使用成员变量yearlyTraffic减少对象创建开销
• 清晰的阶段注释提高代码可读性

3.4 Driver主类配置

public class PhoneTrafficAnalysis { public static void main(String[] args) throws Exception { // 1. 创建配置对象 Configuration conf = new Configuration(); // 2. 创建Job实例 Job job = Job.getInstance(conf, "Phone Traffic Analysis"); job.setJarByClass(PhoneTrafficAnalysis.class); // 3. 设置Mapper/Reducer job.setMapperClass(TrafficMapper.class); job.setReducerClass(TrafficReducer.class); // 4. 指定输入输出类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 5. 设置输入输出路径 FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 6. 提交作业 System.exit(job.waitForCompletion(true) ? 0 : 1); } }

参数说明：

• args[0]：输入文件路径（如hdfs://input/phonetraffic.txt）
• args[1]：输出目录（如hdfs://output/traffic_result）

4. 运行调试与性能优化

写完代码只是开始，如何高效运行和优化才是工程实践的关键。

4.1 本地模式运行

# 打包项目 mvn clean package # 运行Job（本地文件系统） hadoop jar target/your-jar.jar \ PhoneTrafficAnalysis \ file:///path/to/phonetraffic.txt \ file:///path/to/output

常见问题排查：

1. ClassNotFoundException：确保打包时包含依赖
2. 输出目录已存在：手动删除或代码中自动清理
3. 权限问题：检查文件读写权限

4.2 集群模式运行

# 上传数据到HDFS hdfs dfs -put phonetraffic.txt /input/ # 提交作业 hadoop jar target/your-jar.jar \ PhoneTrafficAnalysis \ /input/phonetraffic.txt \ /output/traffic_result # 查看结果 hdfs dfs -cat /output/traffic_result/part-r-00000

4.3 性能优化技巧

1. Combiner优化：

// 在Driver中添加 job.setCombinerClass(TrafficReducer.class);

注意：Combiner和Reducer逻辑相同时可直接复用Reducer类

2. 资源配置：

// 在Driver中调整 conf.set("mapreduce.map.memory.mb", "1024"); conf.set("mapreduce.reduce.memory.mb", "2048");

3. 数据压缩：

// 启用Map输出压缩 conf.set("mapreduce.map.output.compress", "true"); conf.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.SnappyCodec");

基准测试结果对比：

在实际项目中，根据数据特征选择合适的优化组合往往能获得最佳性价比。