Hadoop MapReduce 实战3个经典案例解析学生成绩统计与数据去重在当今数据爆炸的时代处理海量数据已成为企业和技术团队面临的常态挑战。Hadoop MapReduce作为分布式计算的经典框架以其简单而强大的编程模型为处理TB甚至PB级数据提供了可靠方案。本文将深入解析三个典型业务场景通过完整可运行的Java代码示例展示MapReduce在不同数据处理需求中的灵活应用。1. 学生成绩统计找出每位学生的最高分教育数据分析中统计学生成绩是最基础却至关重要的需求。假设我们有一个包含学生姓名和对应分数的文本文件每行记录一次考试结果格式为姓名 分数。我们的目标是找出每位学生所有考试中的最高分。1.1 设计思路这个案例完美契合MapReduce的分组-聚合模式Mapper阶段将每行记录拆解为学生姓名, 分数的键值对Reducer阶段对同一学生的所有分数进行比较保留最大值关键在于Mapper只需原样输出数据而Reducer需要实现简单的比较逻辑。这种设计避免了不必要的数据传输符合MapReduce计算向数据移动的核心原则。1.2 完整代码实现import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class MaxScore { public static class ScoreMapper extends MapperObject, Text, Text, IntWritable { private Text studentName new Text(); private IntWritable score new IntWritable(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String[] record value.toString().split( ); studentName.set(record[0]); score.set(Integer.parseInt(record[1])); context.write(studentName, score); } } public static class MaxScoreReducer extends ReducerText, IntWritable, Text, IntWritable { private IntWritable result new IntWritable(); public void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException { int maxScore Integer.MIN_VALUE; for (IntWritable val : values) { maxScore Math.max(maxScore, val.get()); } result.set(maxScore); context.write(key, result); } } public static void main(String[] args) throws Exception { Configuration conf new Configuration(); Job job Job.getInstance(conf, max score); job.setJarByClass(MaxScore.class); job.setMapperClass(ScoreMapper.class); job.setCombinerClass(MaxScoreReducer.class); job.setReducerClass(MaxScoreReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }1.3 性能优化技巧Combiner的使用本例中Reducer逻辑满足结合律和交换律可以安全使用Combiner在Mapper端先进行局部聚合大幅减少网络传输合理设置Reducer数量根据数据量和集群规模调整避免Reducer过少导致负载不均或过多造成资源浪费数据类型选择使用Hadoop特有的IntWritable而非Java原生int减少序列化开销注意实际部署时应确保输入路径存在且输出路径不存在否则作业会失败。HDFS操作命令如下hdfs dfs -mkdir -p /user/test/input hdfs dfs -put local_file.txt /user/test/input2. 文件内容合并与去重处理多源数据企业数据整合中经常需要合并来自不同系统的数据并去除重复记录。假设我们有两个文件file1和file2每行记录包含学号和课程代码格式为学号 课程。目标是合并两个文件并按学号排序同时去除完全相同的行。2.1 技术难点分析这个案例的特殊性在于需要处理来自多个输入文件的数据去重标准是整行内容完全相同输出需要按学号排序MapReduce的天然分布式特性使其非常适合这种多源数据整合场景。关键在于利用MapReduce的排序机制和Reducer的键唯一性特性。2.2 完整代码实现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.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class MergeAndDeduplicate { public static class MergeMapper extends MapperObject, Text, Text, Text { private Text recordKey new Text(); private Text recordValue new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String line value.toString(); String[] parts line.split( , 2); recordKey.set(parts[0]); // 学号作为键 recordValue.set(parts[1]); // 课程代码作为值 context.write(recordKey, recordValue); } } public static class DedupReducer extends ReducerText, Text, Text, Text { public void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { SetString uniqueCourses new TreeSet(); for (Text val : values) { uniqueCourses.add(val.toString()); } for (String course : uniqueCourses) { context.write(key, new Text(course)); } } } public static void main(String[] args) throws Exception { Configuration conf new Configuration(); Job job Job.getInstance(conf, merge and dedup); job.setJarByClass(MergeAndDeduplicate.class); job.setMapperClass(MergeMapper.class); job.setReducerClass(DedupReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); // 设置多个输入路径 FileInputFormat.addInputPaths(job, args[0] , args[1]); FileOutputFormat.setOutputPath(job, new Path(args[2])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }2.3 关键实现细节多输入处理通过FileInputFormat.addInputPaths()方法指定多个输入路径用逗号分隔TreeSet排序Reducer中使用TreeSet而非HashSet保证课程代码按字母顺序输出键设计以学号作为键利用MapReduce的自动排序特性实现最终结果的学号排序值处理课程代码作为值在Reducer中通过Set自动去重下表对比了不同去重方案的适用场景方案适用场景优点缺点Reducer端Set去重数据量适中重复率高实现简单保证顺序内存消耗大Map端布隆过滤器海量数据预过滤内存效率高存在误判可能二次排序需要复合键去重精确控制实现复杂3. 关系挖掘从父子数据发现祖孙关系社交网络和家族关系分析中经常需要从直接关系中推导间接关系。假设我们有一个child-parent关系表每行记录一对父子关系。目标是挖掘出所有祖孙关系即如果A是B的父母B是C的父母那么A是C的祖父母。3.1 算法设计思路这个案例需要实现类似数据库中的自连接操作技术难点在于需要在Map阶段将每条记录复制为两种视角在Reduce阶段匹配不同视角的记录避免产生无意义的关联解决方案是设计特殊的键值结构让Reducer能够区分不同视角的记录并进行正确关联。3.2 完整代码实现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.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class AncestorFinder { public static class RelationMapper extends MapperObject, Text, Text, Text { public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String[] relation value.toString().split( ); if (relation.length ! 2) return; String child relation[0]; String parent relation[1]; // 作为父节点视角输出 context.write(new Text(parent), new Text(1: child : parent)); // 作为子节点视角输出 context.write(new Text(child), new Text(2: child : parent)); } } public static class RelationReducer extends ReducerText, Text, Text, Text { private static int time 0; public void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { // 输出表头 if (time 0) { context.write(new Text(grandchild), new Text(grandparent)); time; } ListString grandChildren new ArrayList(); ListString grandParents new ArrayList(); for (Text val : values) { String[] parts val.toString().split(:); String relationType parts[0]; String child parts[1]; String parent parts[2]; if (1.equals(relationType)) { grandChildren.add(child); // 作为父节点时其子节点是孙 } else { grandParents.add(parent); // 作为子节点时其父节点是祖 } } // 产生祖孙关系 for (String gc : grandChildren) { for (String gp : grandParents) { context.write(new Text(gc), new Text(gp)); } } } } public static void main(String[] args) throws Exception { Configuration conf new Configuration(); Job job Job.getInstance(conf, ancestor finder); job.setJarByClass(AncestorFinder.class); job.setMapperClass(RelationMapper.class); job.setReducerClass(RelationReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }3.3 执行流程解析Mapper阶段每条child-parent记录被处理两次作为parent视角标记为类型1用于找出该parent的所有child即孙辈作为child视角标记为类型2用于找出该child的所有parent即祖辈Shuffle阶段所有相同key人名的记录会被发送到同一个Reducer包括该人作为parent和作为child的不同视角记录Reducer阶段分离两种类型的记录将类型1的child孙辈与类型2的parent祖辈做笛卡尔积输出所有有效的祖孙关系对这种实现方式巧妙地利用了MapReduce的分组特性避免了显式的自连接操作大大提高了处理效率。对于N条原始关系记录算法时间复杂度为O(N)空间复杂度取决于每个人的关联数量。4. MapReduce优化实战技巧经过三个案例的实践我们总结出以下MapReduce性能优化经验这些技巧在实际生产环境中能显著提升作业执行效率4.1 资源调优配置在job.set方法之后添加以下配置可以优化集群资源利用// 设置Mapper内存 job.getConfiguration().set(mapreduce.map.memory.mb, 2048); // 设置Reducer内存 job.getConfiguration().set(mapreduce.reduce.memory.mb, 4096); // 启用Map输出压缩 job.getConfiguration().set(mapreduce.map.output.compress, true); job.getConfiguration().set(mapreduce.map.output.compress.codec, org.apache.hadoop.io.compress.SnappyCodec);4.2 数据倾斜处理方案当某些键异常频繁时会导致对应Reducer负载过重。解决方法包括预处理采样先运行抽样作业识别热点键增加Reducer数量通过job.setNumReduceTasks()设置自定义Partitioner将热点键分散到多个Reducerpublic class SkewAwarePartitioner extends PartitionerText, Text { Override public int getPartition(Text key, Text value, int numPartitions) { if (key.toString().equals(热点键)) { return (key.hashCode() Integer.MAX_VALUE) % (numPartitions / 2); } return (key.hashCode() Integer.MAX_VALUE) % numPartitions; } }4.3 调试与监控添加计数器帮助定位问题public class MyMapper extends Mapper... { enum Counters { MALFORMED_RECORDS } public void map(...) { try { // 正常处理逻辑 } catch (Exception e) { context.getCounter(Counters.MALFORMED_RECORDS).increment(1); } } }通过Web UI默认端口8088或命令行监控作业状态yarn application -list yarn logs -applicationId app_id4.4 新版本Hadoop特性Hadoop 3.x引入的重要改进特性优势启用方式Erasure Coding节省50%存储空间在hdfs-site.xml中配置GPU调度加速机器学习任务设置mapreduce.gpu.enabled基于cgroup的资源隔离防止任务资源冲突yarn.nodemanager.resource-plugins在实际项目中根据数据规模、集群配置和业务需求选择合适的优化组合通常需要多次测试才能找到最佳配置。