第一次跑通 MapReduce 的作业，大多数人入门都是从 WordCount（也就是标题里的 WordMapper + WordReducer 这套组合）开始的。这个例子看起来简单到有点"不值一提"，但如果你只是把代码抄下来跑完就算了，那基本等于白做。它背后其实藏着一整套分布式计算的通用范式：数据怎么切分、怎么跨节点传输、怎么归并、怎么落盘，搞懂这一条链路，后面再去看 Hive、Spark 甚至 Flink 的很多设计，都会觉得似曾相识。
这篇文章就围绕 WordMapper 和 WordReducer 的代码、编译、打包、运行、排错这条完整链路来写。我会把每一步为什么要这么做讲清楚，而不是只丢一堆命令让你复制。适合刚接触 Hadoop MapReduce、想自己手动写一个作业并跑通的同学，也适合那些跑通过但一直没搞明白"中间到底发生了什么"的人。
1. 别把 WordCount 当成"数单词"的小玩具
1.1 它真正演示的是分布式计算的最小闭环
我第一次接触 WordCount 时也觉得，就这？一个文本里数单词出现的次数，用 Shell 的sort | uniq -c几秒钟就出来了，何必搞一个分布式框架。直到我把数据换成 10 个 GB 的日志文件、放在几十台机器上，才意识到 WordCount 的价值根本不在"数单词"本身，而在于它演示了一个计算任务如何被拆成 N 份并行执行，再汇总成最终结果。
经典问题"如何让 10 个人数完一本书里每个词出现了多少次"和 MapReduce 的思路是完全一致的：每 100 页分给一个人，每个人先数自己负责部分的词频（Map），然后把相同单词的计数归到同一个人手里合并（Shuffle + Reduce），最后把所有小结果拼起来。
放在 Hadoop 里，对应关系是这样：
- Map 阶段：输入文件按块被拆成若干分片，每个分片由一个 Map 任务处理。WordMapper 收到一行文本，把它切成若干个单词，每碰到一个单词就输出一个
<单词, 1>的键值对。 - Shuffle 阶段：框架自动把相同 key 的 value 汇聚到一起，并且key 会按字典序排好，不用你写任何排序代码。
- Reduce 阶段：WordReducer 拿到一个单词和它对应的所有
1，累加得到最终词频，输出<单词, 总次数>。
也就是说，WordCount 完整覆盖了 MapReduce 的map -> shuffle -> reduce -> 输出四个环节。你把这个环节里的每个数据结构、每个方法的参数搞懂，就等于拿到了一把钥匙，所有 MapReduce 作业的骨架都在这里。
1.2 输入格式与输出格式：别在格式上走弯路
MapReduce 默认的输入格式是TextInputFormat，它把文件按行切分，每一行交给 Mapper，并且生成一个 key-value 对：
- key是该行在整个文件中的偏移量（字节数），类型是
LongWritable - value是这一行的文本内容，类型是
Text
这也是为什么 WordMapper 的泛型签名必须是Mapper<LongWritable, Text, Text, IntWritable>的原因。有同学问过，为什么第一个参数不能用Integer、第二个不能用String？原因在于 Hadoop 的分布式序列化机制。Java 自带的序列化会把整个对象图写进字节流，冗余信息太多，所以 Hadoop 定义了自己的序列化类型：LongWritable、Text、IntWritable这些 Writable 类型体积小、可高效序列化，在跨网络传输时开销远低于原生 Java 类型。这是初学者最容易忽略但面试也最常被问到的点。
输出格式默认是TextOutputFormat，Reducer 每context.write(key, value)一次，输出文件里就多一行key + Tab + value。最终文件会以part-r-xxxxx命名，r表示 reducer 的输出，xxxxx是 reducer 的编号。看到这个文件，说明整个作业成功了。
2. 逐行拆解：WordMapper、WordReducer 和驱动类
2.1 WordMapper：map 方法输入的 KV 对从哪来
先直接看代码，我用的是 Hadoop 2.x / 3.x 通用的org.apache.hadoop.mapreduce包（注意不是老的mapred包，新版 API 写起来更简洁，社区也都用它）：
import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; import java.util.StringTokenizer; public class WordMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private final static IntWritable ONE = new IntWritable(1); private Text word = new Text(); @Override protected void map(LongWritable 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); } } }关键点有几个：
第一，value 的类型是Text，但StringTokenizer要接收String，所以要做一次toString()。Text和 JavaString虽然长得很像，但不是同一个类型，不能直接互相赋值。
第二，word复用了同一个对象。这里有个很微妙的细节：在循环里对同一个word对象反复set()，然后context.write(word, ONE)，看起来每次都写了"同一个对象"，会不会导致最终所有输出都变成最后一个单词？答案是不会。context.write会把当前的 key 和 value 序列化后放入缓冲区，而不是持有这个对象的引用，所以复用对象是安全的，而且能减少对象创建的开销，是官方推荐写法。
第三，ONE是静态常量。每个单词的初始计数都是 1，不需要每次 new 一个新的IntWritable。
第三点可以展开说一下：为什么不能直接写context.write(word, new IntWritable(1))？能运行，但 map 任务可能处理百万行、千万行记录，每次都创建对象，GC 压力会明显上升。用静态复用的 Writable 对象，是 MapReduce 性能优化里"减少对象创建"这个原则最直接的体现。
2.2 WordReducer：reduce 方法拿到的 values 是已经聚合好的迭代器
Reducer 的代码同样不长：
import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class WordReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } }这里初学者最容易犯的错是：想把values转成List再遍历。不要这么干。values是一个延迟迭代的Iterable，它背后可能是一大堆来自不同 Map 任务、通过网络传输过来的数据。如果数据量大，你把它全部缓存在内存里，轻则内存溢出，重则拖垮整个 NodeManager。正确做法就是在这里用一次遍历直接聚合完。
Reducer 的输入是已经排好序的：同一个 key 的所有 value 会被分成一组，调用一次reduce()，组的顺序按照 key 排序。也就是说，如果你有 10 万个单词，reduce()方法会被调用 10 万次，每次处理一个单词的所有计数。
另外注意输出的 key 类型是Text，value 类型是IntWritable，这和 Mapper 的输出类型是一致的。Shuffle 阶段搬运的就是 Mapper 的<key, value>对，所以Mapper 的输出 key/value 类型，必须和 Reducer 的输入 key/value 类型完全匹配，否则运行到一半会报类型转换错误。
2.3 驱动类：配置 Job 的三大核心动作
有了 Mapper 和 Reducer，还需要一个驱动类把它们串起来提交给集群。这也是"WordReduce"这类描述里最容易漏掉的部分——很多人以为写完 Mapper 和 Reducer 就完了，实际上没有 driver，Hadoop 根本不知道你写的是什么。
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.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class WordCountDriver { public static void main(String[] args) throws Exception { if (args.length != 2) { System.err.println("Usage: WordCountDriver <input path> <output path>"); System.exit(-1); } Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "Word Count"); job.setJarByClass(WordCountDriver.class); job.setMapperClass(WordMapper.class); job.setReducerClass(WordReducer.class); job.setCombinerClass(WordReducer.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); } }这里的核心逻辑只有三件事：
- 设置 Jar 入口类：
setJarByClass(WordCountDriver.class)。Hadoop 需要知道从哪个类加载出包含 Mapper 和 Reducer 的 Jar 包。如果不设置，在集群模式下经常报ClassNotFoundException。 - 注册 Mapper 和 Reducer 类：让框架知道作业的 map 逻辑和 reduce 逻辑分别是哪个类。
- 声明输出 key/value 类型：
setOutputKeyClass和setOutputValueClass设置的是Reduce 阶段的输出类型。如果 Mapper 的输出类型和 Reducer 的输出类型不一致，还需要额外调用setMapOutputKeyClass和setMapOutputValueClass。
我先说结论：只要 Mapper 的输出类型和 Reducer 的输出类型一致，上面这段代码就够了。因为默认情况下 Hadoop 会认为 map 输出的 key/value 类型和 job 的输出类型相同。如果哪天你的 Mapper 输出的是<Text, LongWritable>，Reducer 最终输出的是<Text, IntWritable>，就必须显式补充setMapOutputKeyClass(Text.class)和setMapOutputValueClass(LongWritable.class)，否则运行到 map 输出阶段就会因为类型不匹配而报错。
还有一个小细节：job.setCombinerClass(WordReducer.class)这一行的注释我特意保留了。Combiner 是可选项，但它对性能影响极大，后面第五章会专门讲。这里先记住一个原则：Combiner 的逻辑必须满足交换律和结合律，否则结果会出错。词频累加天然满足，所以可以直接复用 Reducer 类。
3. 从代码到集群运行：编译、打包、提交的完整链路
3.1 Maven 依赖与打包注意事项
MapReduce 程序要用 Maven 管理依赖，最核心的依赖就一个：
<dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.3.6</version> <scope>provided</scope> </dependency>scope设为provided很重要，因为你的集群环境里已经有一整套 Hadoop 库了，不需要把 Hadoop 的依赖打进 Jar 包。如果把 Hadoop 所有依赖都打进去，生成的 Jar 可能几十兆，提交作业时网络传输慢不说，还很可能和集群自带的类产生冲突。
打包命令用最朴素的 Maven 方式即可：
mvn clean package -DskipTests如果你没有用 Maven，而是手工javac编译，那需要把集群上的hadoop-classpath加到编译路径里，然后把编译好的 class 文件打成 Jar。这种纯手工方式在只有一个 class 文件时够用，但项目一复杂就容易乱，还是建议用 Maven，一步到位。
打包后确认一下 Jar 里的类是不是齐全：
jar tf target/wordcount-demo-1.0.jar | grep Word正常情况下应该能看到WordMapper.class、WordReducer.class、WordCountDriver.class三个 class 文件。
3.2 输入数据准备：先上传到 HDFS
MapReduce 读取的不是本地文件系统的文件，而是HDFS 上的文件。这一步是很多初学者在本地试运行时报"文件找不到"的根源。
准备一个测试文件，内容随便写几行英文，例如：
hello world hadoop hello mapreduce hadoop mapreduce wordcount demo然后上传 HDFS：
hdfs dfs -mkdir -p /user/yourname/input hdfs dfs -put words.txt /user/yourname/input/读取时路径可以写相对路径：
hadoop jar target/wordcount-demo-1.0.jar WordCountDriver input output这里input会解析为你当前用户在 HDFS 家目录下的/user/yourname/input。如果你不确定，建议先用绝对路径/user/yourname/input跑一次，跑通了再换成相对路径。
3.3 提交作业与运行日志的关键行
用hadoop jar提交作业后，控制台会滚动输出进度。你需要关注的几个关键信息：
Number of maps：Map 任务数量。一个输入分片对应一个 Map 任务，默认分片大小和 HDFS 块大小一致。对于小文件，一个文件可能就是一个 Map 任务。Number of reduces：默认是 1，也就是一个 Reduce 任务。map 100% reduce 100%：看到这两个 100% 基本就稳了。File System Counters和Map-Reduce Framework里能看到输入记录数、输出记录数，可以用来校验结果是否符合预期。
跑完以后查看结果：
hdfs dfs -cat /user/yourname/output/part-r-00000输出类似：
hadoop 2 hello 2 mapreduce 2 world 1 wordcount 1 demo 1如果你在集群的 ResourceManager Web 界面（默认端口 8088）能看到作业历史记录，还可以点进去看每个 Map/Reduce 任务的运行时长、GC 时间、Shuffle 字节数等指标。这些指标在调优时非常有用。
4. 跑起来以后的高频报错：从现象到根因的排查链路
4.1 输出目录已存在：最让新手困惑的"不是 bug 的 bug"
第一次提交作业，大概率会遇到这一行：
org.apache.hadoop.mapreduce.lib.output.FileOutputFormat$FileOutputCommitter: FileAlreadyExistsException: Output directory output already exists原因很简单：Hadoop 为了安全，拒绝覆盖已有的输出目录，以免你上一次运行的结果被悄悄冲掉。这也意味着你每跑一次新的作业，都要删掉旧的输出目录或者换一个新名字：
hdfs dfs -rm -r /user/yourname/output或者直接换一个路径output2。我建议养成"每次运行换新目录"的习惯，因为某些时候你希望对比两次运行的结果，旧目录留着反而方便。
4.2 用户权限问题：本地能跑，集群却写不进去
本地跑伪分布式或单机模式时极少遇到权限问题，一旦上了多节点集群，各种Permission denied就开始冒出来了。常见场景是：你用 A 用户提交作业，但输出目录的父目录是 B 用户创建的，你没有写权限。
解决方式有两种：
- 让管理员给输出目录授权，或者你在 HDFS 上建一个自己有权限的目录做测试。
- 如果只是个人测试环境，可以临时用
hdfs dfs -chmod -R 777 /user/yourname/output放宽权限，但生产环境不要这么干。
还有一类权限问题容易迷惑人：报错提示PriviledgedActionException或者AccessControlException，它可能不是你操作的问题，而是集群的dfs.permissions.enabled开启了严格权限校验。这时候优先看完整堆栈里到底卡在哪个路径上，用hdfs dfs -ls逐级检查该路径的 owner 和权限即可。
4.3 ClassNotFoundException：Jar 没打全或不认识入口类
这个报错也有几种不同表现：
- 报
ClassNotFoundException: WordMapper：多半是 Jar 里根本没这个 class，或者你提交作业时指定的入口类加上了包名但实际没有包。使用jar tf检查一下。 - 报
ClassNotFoundException: org.apache.hadoop.mapreduce.Mapper：说明你运行环境的HADOOP_CLASSPATH没配置好，或hadoop命令找不到集群的库。检查hadoop classpath命令输出是否正常。 - 报入口类不对：确认你提交时的主类名，如
WordCountDriver，而不是 Mapper 或 Reducer 类。有同学直接把WordMapper作为主类提交，Hadoop 进去了发现没有 main 方法，报错明明白白，但你第一眼不一定反应过来。
还有一个隐藏很深的坑：用 IDE 写代码编译运行没问题，但打成 Jar 时把hadoop-client依赖也打进去了，导致集群上类冲突、行为怪异。解决办法就是前面说的，把依赖 scope 设置为provided。
4.4 本地调试的小技巧：先用单机模式快速跑通
线上集群排错成本高，我一般先在本地用单机模式跑通逻辑，再提交集群。做法是直接在 IDE 或命令行用原生的输入输出路径来执行：
hadoop jar target/wordcount-demo-1.0.jar WordCountDriver \ file:///tmp/words.txt \ file:///tmp/wordcount-out注意路径前缀file://，表示读本地文件而不是 HDFS。只要 Hadoop 环境变量（HADOOP_HOME）配好，本机就能直接跑。这种方式特别适合验证 Mapper/Reducer 逻辑本身是否正确，排除集群环境的干扰。
如果连本机 Hadoop 都不想装，另一个轻量方案是用 MRUnit 写单元测试。Mock 一个 Map 输入，断言输出 KV 是否符合预期：
@Test public void testWordMapper() throws Exception { new MapDriver<LongWritable, Text, Text, IntWritable>() .withMapper(new WordMapper()) .withInput(new LongWritable(1), new Text("hello world")) .withOutput(new Text("hello"), new IntWritable(1)) .withOutput(new Text("world"), new IntWritable(1)) .runTest(); }MRUnit 的依赖只在测试时使用，不影响最终 Jar，是快速验证 Mapper/Reducer 逻辑的好工具。
5. 从 WordCount 走向实战：Combiner、自定义类型与调优思路
5.1 Combiner：数十倍 Shuffle 流量削减的秘密
回到驱动类里被很多人忽略的job.setCombinerClass(WordReducer.class)。Combiner 的作用是在 Map 端先做一次局部归并，减少要传输到 Reducer 的数据量。
用一个例子说明：假设某个 Map 任务处理的数据里有 10 万次hadoop这个单词。没有 Combiner，它要向 Reducer 发送 10 万个<hadoop, 1>；有 Combiner，它在本地先累加，只发一个<hadoop, 100000>。Shuffle 阶段传输的数据量可能直接缩小一个数量级，整个作业的完成时间也随之缩小。
但 Combiner 不是想用就能用的。同一个 Combiner 可能会被调用多次，且不保证每次处理的是完整的一组 value，所以它必须满足交换律和结合律。词频累加满足；但如果你做的是求平均值，在 Combiner 里算均值就完全错误。求均值要做 Combiner，必须用"总和 + 个数"这种复合数据结构，这就是为什么会有自定义 Writable 的经典场景。
注意：Combiner 的输入输出类型必须和 Mapper 的输出类型一致，这样它才能被插在 Shuffle 链路中间。如果你复用了
WordReducer，它的输入输出都是<Text, IntWritable>，正好满足这个约束。
5.2 自定义 Writable：计数器、日志统计的进阶形态
WordCount 的 value 是IntWritable，数据类型太简单，实战里根本不够用。比如要统计"每个单词在每台服务器上出现的总次数和平均长度"，一个 value 就装不下两个字段。
自定义类型标准的做法是继承Writable，实现write和readFields两个方法，并保证序列化和反序列化的字段顺序一致：
import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class WordStatsWritable implements Writable { private long count; private long totalLength; public WordStatsWritable() {} public WordStatsWritable(long count, long totalLength) { this.count = count; this.totalLength = totalLength; } @Override public void write(DataOutput out) throws IOException { out.writeLong(count); out.writeLong(totalLength); } @Override public void readFields(DataInput in) throws IOException { this.count = in.readLong(); this.totalLength = in.readLong(); } public long getCount() { return count; } public long getTotalLength() { return totalLength; } }注意必须保留一个无参构造器，因为 Hadoop 反序列化时需要反射调用它，没有会直接报NoSuchMethodException。
这里引出一个实战中很常见的需求：Reducer 里如果要输出多个维度（比如词频结果和总字数统计），可以用MultipleOutputs实现，它允许你按"命名前缀"写多个输出文件，避免把所有结果塞到一个part-r-00000里。这个 API 的用法不复杂，但非常实用，尤其是在做数据清洗时想把正常数据和异常数据分开落盘。
5.3 分区与二次排序：真正理解 Shuffle 语义的分水岭
很多人以为 Reducer 收到的 value 顺序是可用的，但其实框架只保证 key 有序，同一 key 内部 value 的顺序默认是不确定的。如果你想实现"按 key 分组，组内按时间排序"的效果，就用到二次排序（Secondary Sort）机制：自定义一个组合 key，让它既携带分组字段又携带排序字段，再通过自定义Partitioner和GroupingComparator控制分组逻辑。
WordCount 用不到这么复杂的东西，但它是理解 Shuffle 内部机制的最佳切入点。理解了它，你再看 Hive 的SORT BY、DISTRIBUTE BY，就会明白它们不过是映射到了 MapReduce 的排序和分区阶段。
分区方面，如果你想让"相同首字母的单词进入同一个 Reducer"，可以实现一个自定义Partitioner<Text, IntWritable>：
@Override public int getPartition(Text key, IntWritable value, int numPartitions) { if (key.toString().isEmpty()) return 0; return Math.abs(key.toString().charAt(0) - 'a') % numPartitions; }然后在驱动类里：
job.setPartitionerClass(FirstLetterPartitioner.class); job.setNumReduceTasks(5);设置多少个 Reducer，最终就会生成多少个part-r-xxxxx文件。注意，Reducer 数量设置不合理是性能杀手：设得太多，每个 Reducer 处理的数据太少，启动和调度开销占比高；设得太少，所有数据压在少数几个节点上，又变成木桶效应。一般经验值是每个 Reducer 处理 1 GB 左右的数据，或者按集群 CPU 核数综合考虑。
5.4 参数调优：从"能跑"到"跑得快"
最后分享几个我平时调试 MapReduce 作业时最常用的调优点，不是大而全的参数手册，而是对 WordCount 这类作业最立竿见影的几个：
第一，调整 Map 和 Reduce 内存。mapreduce 容器默认内存只有 1 GB 左右，如果你的每个 Map 任务要处理的数据结构占内存较大，可以在提交时指定：
hadoop jar target/wordcount-demo-1.0.jar WordCountDriver \ -Dmapreduce.map.memory.mb=2048 \ -Dmapreduce.reduce.memory.mb=3072 \ input output第二，压缩输出数据。HDFS 上存文本文件非常浪费空间，MapReduce 输出可以启用压缩：
-Dmapreduce.output.fileoutputformat.compress=true \ -Dmapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.SnappyCodecSnappy 压缩速度快、压缩比适中，是日志和数据交换场景的常见选择。注意压缩后的文件不能直接hdfs dfs -cat查看，需要配合zcat或 Hadoop 自带的文本查看工具。
第三，小文件问题。如果你的输入目录里有大量几 KB 的小文件，MapReduce 会给每个文件启动一个 Map 任务，任务数暴涨但每个任务干一点点活，整体效率极低。解决办法是先用CombineTextInputFormat把小文件合并成分片：
job.setInputFormatClass(CombineTextInputFormat.class); CombineTextInputFormat.setMaxInputSplitSize(job, 67108864); // 64 MB这个改动能把成千上万个 Map 任务压到几十个，效果立竿见影。数据分析领域有句话叫"小文件是 HDFS 的癌症"，MapReduce 场景里体现得最明显。
我在实际配置集群作业时还有一个体会：先跑一个小的样本数据，观察日志里的Number of splits和每个任务的输入数据量，再决定要不要调分片大小和 Reducer 数量。一上来就套用别人给的参数，往往适得其反。你用的数据规模、集群核数、内存大小和别人不一样，最优参数就不可能一样。
最后再补充一个值得记住的经验：WordCount 虽然简单，但它是极少数能让你完整看到Map -> Shuffle -> Reduce -> 输出全链路的例子。如果你在学习阶段，建议把它改成不同的输入格式（比如 JSON 日志、CSV 文件）反复运行，观察分片数量、排序行为、Combiner 生效与否对指标的影响。把这些体感建立起来之后，再看任何复杂的分布式计算框架，你都会比那些只背概念的人多一层"它底层到底怎么跑"的直觉。
