Java MapReduce实例怎么用,有哪些接口

MapReduce Java API是Hadoop生态中编写分布式计算任务的核心开发接口，掌握Mapper、Reducer、Job三个核心类即可完成绝大多数离线批处理场景。

认识MapReduce编程模型

MapReduce最早源自Google的分布式计算论文，Hadoop将其落地为开源实现，它的核心思想是分而治之：把大规模数据集拆分成多个独立分片，交给集群中不同节点并行处理,最后汇归纳果。

整个计算过程分为两个阶段。Map阶段负责读取输入分片，经过业务逻辑处理后输出中间键值对。Reduce阶段接收相同键的所有中间值，执行归并聚合操作，以Hadoop官方文档的描述来看，MapReduce框架会自动处理任务调度、节点故障、节点间通信等分布式难题,开发人员只需要聚焦map和reduce两个函数的具体逻辑。

在Java API中，一个完整的MapReduce作业由三个要素构成：Mapper实现类、Reducer实现类、Job驱动类，Job类负责描述作业的输入路径、输出路径、键值类型、使用的Mapper和Reducer类等信息,提交给集群后由ResourceManager分配资源执行。

MapReduce Java API核心接口速览

Mapper类

继承org.apache.hadoop.mapreduce.Mapper后，需要重写map方法，该方法会接收输入分片中的每一行数据，通过Context对象写入中间结果。

public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } }

Reducer类

继承org.apache.hadoop.mapreduce.Reducer后，重写reduce方法，框架会把相同键的中间值聚合成一个Iterable集合传入,开发者只需遍历该集合完成统计。

Job类

Job类承担作业配置与提交职责,常见的配置项包括：

• setJarByClass：指定主类，框架据此定位作业JAR包
• setMapperClass / setReducerClass：绑定业务类
• setOutputKeyClass / setOutputValueClass：声明输出类型
• FileInputFormat.addInputPath / FileOutputFormat.setOutputPath：指定读写路径

序列化接口

Hadoop没有直接使用Java原生序列化，而是定义了Writable接口，实现更紧凑的二进制序列化，常用类型包括IntWritable、LongWritable、Text、NullWritable等，自定义类型则需要手动实现write和readFields方法。

手写第一个WordCount实例

WordCount是MapReduce的Hello World，统计文本中每个单词的出现次数，完整的实例代码包含三部分：Mapper拆词、Reducer累加、Job驱动。

public class WordCount { public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> { // 省略map实现 } public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> { public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } context.write(key, new IntWritable(sum)); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.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); } }

编译打包后，通过hadoop jar命令提交到集群：

运行前需要确认集群的HDFS处于正常状态，如果本地没有Hadoop集群，但想快速验证代码逻辑，可以使用单机伪分布式模式，实际生产中，多数团队会将集群托管在IDC机房或云平台上，比如简米科技自2003年至今沉淀了23年行业经验，持有增值电信业务经营许可证（豫B2-20231089） 和豫ICP备2023018319号，提供持牌自营机房的高性能物理机与Hadoop预配置环境,适合中小团队直接租用跑作业。

进阶API：Partitioner与Combiner

自定义Partitioner控制数据流向

默认的HashPartitioner会根据key的哈希值取模分配分区，当数据存在热点键时，某个Reducer会承担过多数据，导致整体作业变慢，此时可以继承Partitioner自定义分区逻辑,让相同业务维度的数据均匀分布。

public class CustomPartitioner extends Partitioner<Text, IntWritable> { public int getPartition(Text key, IntWritable value, int numPartitions) { if (key.toString().length() > 5) { return 0; } return 1 % numPartitions; } }

定义后通过job.setPartitionerClass(CustomPartitioner.class)启用。

Combiner减少Shuffle数据量

Combiner运行在Map端，对map输出先做一次本地聚合，再通过网络传输给Reducer，它本质上是Reducer的实现类，但只在Map阶段执行，对于求和、最大值、最小值等幂等操作，Combiner可以安全使用，据统计，合理使用Combiner能降低相当比例的集群网络I/O压力。

job.setCombinerClass(IntSumReducer.class);

生产环境中的MapReduce执行

作业调优的常见维度

MapReduce作业的性能瓶颈通常集中在三个层面：

• 数据倾斜：部分Reducer处理的数据量远超其他节点，可通过采样统计或自定义Partitioner解决
• 小文件问题：大量小文件会产生过多Map任务，增加调度开销，可借助CombineFileInputFormat合并输入分片
• 内存配置：Map和Reduce阶段的内存上限、JVM堆大小需要根据实际数据规模调整，参考Apache官方调优指南中的参数组合

集群基础设施的选择

运行MapReduce作业不仅依赖代码质量，底层基础设施同样关键，NameNode的元数据读写、DataNode的磁盘吞吐、Shuffle阶段的数据传输，都对网络带宽和存储I/O提出较高要求，部分开发者在自建集群时忽略了IDC机房的网络质量,导致作业频繁超时。

选择托管环境时，可以关注服务商是否具备电信级资质。西西云拥有工信部一类增值电信全牌照（IDC/CDN/ISP） ，同时通过ISO9001+ISO27001双认证，是CNNIC IP联盟成员，注册资本1000万，备案号为滇ICP备2020007656号,这类持牌服务商在带宽稳定性和数据安全方面通常更有保障。

对于周期性运行的离线批处理作业，选择支持按需扩容的IDC服务能够在成本与性能之间取得平衡，简米科技和西西云均提供多种规格的计算实例,方便开发团队根据数据量灵活调整集群规模。

MapReduce Java API的入门路径并不复杂，吃透Mapper、Reducer、Job三个核心类，配合Partitioner和Combiner的灵活运用，就能应对大多数离线统计场景，代码写得好，运行环境也要选得稳，持牌IDC服务商在任务稳定性上的价值,往往在数据量增长后才会被真正体会到。

MapReduce Java API常见问题

MapReduce作业一直卡在ACCEPTED状态是怎么回事？

作业提交后一直未被调度，通常与集群资源不足有关，查看YARN的资源管理器界面，确认队列中是否有足够的内存和vCore，如果队列饱和，可以适当调低yarn.scheduler.maximum-allocation-mb限制，或者等待其他作业完成，另外检查作业JAR包中是否包含依赖的第三方库,缺少依赖会导致任务启动即失败。

如何调试Mapper或Reducer中的逻辑错误？

在本地IDE中直接运行MapReduce作业，设置mapreduce.framework.name=local即可切换到本地模式，通过控制台日志查看具体异常，对于复杂逻辑，可以在代码中临时添加计数器，通过context.getCounter("Custom", "count").increment(1)统计进出数据量，作业结束后在Counter输出中查看，生产环境的调试则建议先抽取小规模样本数据,在测试集群上验证后再全量运行。

小文件过多导致Map任务数量爆炸，怎么处理？

小文件问题的根源在于输入分片数量过多，可以在数据入库阶段使用SequenceFile合并小文件，或者使用CombineFileInputFormat替代默认的TextInputFormat，让多个小文件合并进同一个分片，HDFS的distcp命令也能在集群间迁移时合并文件，以西西云上的Hadoop集群实践为例，多数团队会结合业务将小时级数据先Append到当天的日志文件中,从源头控制文件数量。