刚接触Hadoop的人看到“wordMapper”“wordReduce”这种叫法，其实对应的就是MapReduce里最经典的WordCount程序：统计一堆文本中每个单词出现了多少次。它看起来简单，但如果你能亲手把这段代码写完、编译、打包、提交到集群上跑通，MapReduce的核心机制基本就打通了一半。这篇文章我按照平时真实跑任务的顺序来写，从Mapper和Reducer怎么设计与实现，到用什么命令提交、怎么读日志、踩了哪些坑，一次讲透。适合第一次接触Hadoop、想把MapReduce程序真正跑起来的读者。
1. 先想清楚：WordCount的“Map+Reduce”到底在拆解什么
1.1 为什么需要拆成“Mapper + Reducer”两个阶段
单机统计单词这事太简单了：读一个文件，按空格切词，放进一个Map里累加就行。可一旦文件大到几十GB甚至几TB，一台机器装不下，也处理不完，就需要把任务拆开，交给多台机器同时算。MapReduce的聪明之处在于，它把任何一个“分组统计”类任务都抽象成了两个阶段：
- Map阶段：负责“拆”。把原始数据拆成一条一条的中间记录。
- Reduce阶段：负责“合”。把相同key的中间记录汇总成最终结果。
用图书馆整理图书来类比。一大堆书散在各个楼层的书架上（输入数据），你不可能让一个人把所有书数一遍。先派一批人去不同楼层，每人只管自己负责的架子，把每本书按类别抄在纸条上（Map）。然后这些纸条按类别集中到对应柜台（Shuffle），最后由柜台工作人员把每类书的数量汇总（Reduce）。整个过程不需要一个人看到全部数据，但最终结果却和全局统计一模一样。
WordCount就是这种思路最小的载体。Mapper把每个单词变成“word, 1”这种记录，Reducer把相同单词后面的1全部加起来，就是最终次数。
1.2 核心流程不只是Map和Reduce，还有中间的Shuffle
很多人以为MapReduce就是写两个函数，其实真正的工作量在Map和Reduce之间的Shuffle阶段。一次完整任务包含这样几个环节：
- 输入分片：输入文件被切成多个split，一个split对应一个MapTask。
- Map跑完：输出结果先落到本地磁盘，并经过分区、排序、溢写、合并。
- 拉取：ReduceTask从各个MapTask所在节点拉取属于自己分区的数据。
- 归并：Reduce端把拉过来的数据进行再次排序和合并。
- Reduce函数执行：把同key的一组value传给你的reduce方法。
- 输出到HDFS：每个ReduceTask写一个结果文件。
Shuffle是MapReduce运行的骨架，理解它之后，再看代码你会觉得写两个类只是把面条下锅，真正决定这碗面好不好吃的是煮面的过程。
1.3 WordCount到底能迁移到哪些真实场景
很多初学者以为WordCount只是教学玩具，其实它是大量离线统计任务的模板。日志分析里统计某个接口被调用了多少次、爬虫里统计不同域名下的链接数量、内容分发里统计热门关键词的排名，本质上都是“分组计数”。把输入从“单词”换成“IP”“用户ID”“商品编号”，把Reducer里的累加换成计数、求和、求最值，就是一个新的MapReduce应用。所以我一直建议新人不要跳过这个程序，它值得你反复跑通、反复拆解。
2. 环境准备：不搭集群也能把MapReduce完整跑起来
2.1 别一上来就搞分布式集群，先跑伪分布式
很多人第一次学Hadoop就想去申请几台服务器搭集群，结果卡在节点互通、免密登录、Namenode起不来这些环境问题上，连代码都没来得及写。我强烈建议先在单机伪分布式模式下跑通WordCount。
伪分布式什么意思？就是一台机器同时充当Namenode、Datanode、ResourceManager、NodeManager。从代码角度来说，你编写的Mapper、Reducer、Driver在这个模式下运行的流程，和真实集群几乎完全一致，唯一区别是只有一台机器。好处是：
- 环境搭建成本低，配置简单，适合验证逻辑。
- 对学习Shuffle、日志、任务调度没有理解损失。
- 之后上真实集群，只需要调整地址和权限，代码基本不用改。
先把“程序能跑通”这件事解决，再谈“机器变多了怎么办”。
2.2 Hadoop版本和JDK版本怎么搭配
版本搭配是所有环境问题里最容易让人崩溃的。Hadoop 2.x时代用JDK8，Hadoop 3.x也支持JDK8，但部分新功能需要JDK11。我的建议是选择一个稳定组合：Hadoop 3.3.x 配 JDK8 或 JDK11，这是目前最常用、遇到问题最容易搜到解决方案的组合，不要一味追新。
推荐使用Hadoop 3.3.6这类成熟版本。太老的2.7系列虽然教程多，但和现代操作系统的兼容性越来越差；太新的版本可能和某些生态组件还没适配。稳定压倒一切。
2.3 解压、配置环境变量、准备启动脚本
第一步下载Hadoop压缩包，解压到某个目录，假设是/opt/hadoop：
tar -zxvf hadoop-3.3.6.tar.gz -C /opt/ mv /opt/hadoop-3.3.6 /opt/hadoop然后编辑/etc/profile或~/.bashrc，配置环境变量：
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64 export HADOOP_HOME=/opt/hadoop export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin配置$HADOOP_HOME/etc/hadoop/hadoop-env.sh，把JAVA_HOME写死。很多启动失败的根本原因就是这一步没配置好：
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64接着是三个核心配置。core-site.xml里设置NameNode的通信地址和临时目录：
<configuration> <property> <name>fs.defaultFS</name> <value>hdfs://localhost:9000</value> </property> </configuration>hdfs-site.xml里设置副本数，伪分布式下副本数为1就够了：
<configuration> <property> <name>dfs.replication</name> <value>1</value> </property> </configuration>yarn-site.xml里启用YARN资源调度，并设置NodeManager的辅助服务：
<configuration> <property> <name>yarn.nodemanager.aux-services</name> <value>mapreduce_shuffle</value> </property> <property> <name>yarn.nodemanager.aux-services.mapreduce_shuffle.class</name> <value>org.apache.hadoop.mapred.ShuffleHandler</value> </property> </configuration>之后格式化NameNode并启动所有进程：
hdfs namenode -format start-dfs.sh start-yarn.sh用jps命令能看到五个关键进程：NameNode、DataNode、ResourceManager、NodeManager，以及（如果启动了作业历史服务器）JobHistoryServer。这一步过了，环境基本靠谱。
2.4 验证环境是否可用
在HDFS上建一个测试目录，上传一个文本文件，先确认分布式文件系统本身没问题：
hdfs dfs -mkdir -p /user/demo/input echo "hello world hello hadoop" > input.txt hdfs dfs -put input.txt /user/demo/input hdfs dfs -cat /user/demo/input/input.txt能看到文件内容并且不会报权限错误，说明HDFS读写没问题，可以进入写代码环节了。
3. 编程代码：手写WordCount的Mapper、Reducer和Driver
3.1 Mapper代码逐段拆解
在项目目录里新建三个Java文件，我习惯用默认包，方便命令行直接编译和提交。第一个文件是 WordCountMapper.java。
import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private static final IntWritable ONE = new IntWritable(1); private Text word = new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); String[] words = line.split(" "); for (String w : words) { word.set(w); context.write(word, ONE); } } }泛型四项分别对应：输入key类型（行偏移量）、输入value类型（一行文本）、输出key类型（单词）、输出value类型（计数1）。Hadoop不直接使用Java的String和Integer，而是用Text和IntWritable，因为Hadoop需要对象具备高效的序列化能力，才能在网络和磁盘之间传输。
map方法里做的事情就是：取一行文本，按空格切分，把每个单词写成“word, 1”。这里有一个小细节需要注意：我用了同一个Text对象word，每次通过set方法覆盖旧值。这是Hadoop官方示例的推荐写法，因为Context内部只会引用这个对象做后续处理，重复创建对象会增加GC压力。
3.2 Reducer代码逐段拆解
第二个文件是 WordCountReducer.java。
import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class WordCountReducer 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); } }在reduce方法里，同一个单词所有的value会被放在一个Iterable里，你不需要关心它们在哪些节点计算过、怎么传输过来的，框架已经帮你归并好了。你只需要遍历values累加，然后输出单词和总次数。
写到这里，代码逻辑完全对称：map把一个单词拆成多条记录，reduce把多条记录合成一条结果。这就是WordCount的骨架。
3.3 Driver主类，五步设置到位
第三个文件是 WordCountDriver.java，负责组装整个任务。
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 <inputPath> <outputPath>"); System.exit(2); } Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordCountDriver.class); job.setMapperClass(WordCountMapper.class); job.setReducerClass(WordCountReducer.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); } }Driver里有五步缺一不可：
- 用Configuration创建Job实例。
- 设置Jar入口类（setJarByClass），框架靠这个找到包含所有类的jar包。
- 设置Mapper和Reducer类。
- 设置输出key和value的类型。这一步很关键，如果漏掉或设错，运行时会因为序列化类型不匹配直接报错。
- 设置输入路径和输出路径。
如果你的Mapper输出类型和最终输出类型不一样，还需要单独调用job.setMapOutputKeyClass和job.setMapOutputValueClass。WordCount场景两者一致，所以只需要设置最终输出类型。
3.4 把数据流在脑子里走一遍
假设输入文件里有一行：
hello world hello hadoopMap阶段处理之后，会输出：
hello, 1 world, 1 hello, 1 hadoop, 1经过Shuffle分组，Reducer会收到这样的数据：
hello, [1, 1] world, [1] hadoop, [1]Reduce方法把每个列表里的1加总，输出：
hadoop 1 hello 2 world 1注意结果文件中key的顺序不是你输入的顺序，而是Shuffle过程中按key排序后的顺序。这也是MapReduce的一个特点：不保证输出顺序和输入一致，但保证相同key一定被分到同一个Reducer。
4. 编译打包与运行：从javac到hadoop jar的完整命令
4.1 为什么先用javac直接编译，再打jar包
很多新人在IDE里写完代码，点一下运行以为就完事了。但MapReduce程序是在分布式框架里运行的，必须把class文件打成jar包，再由hadoop命令提交。理解这条链路，比会用IDE里的“Run”更重要。
在源码目录下，先编译。Hadoop自带全套依赖，通过hadoop classpath可以直接拿到：
javac -classpath "$(hadoop classpath)" -d wc_classes WordCountMapper.java WordCountReducer.java WordCountDriver.java-d wc_classes表示把class文件输出到wc_classes目录。之后打进jar包：
jar -cvf wordcount.jar -C wc_classes/ .到此你已经得到了一个可提交的wordcount.jar。有些教程强调要用Maven或Gradle打包，但如果你只是想跑通一个MR程序，命令行原生的方式反而更透明，出错了也更容易定位是哪一步的问题。当然，如果项目里依赖很多，Maven是更好的选择，但WordCount这种三个Java文件的程序没必要上重量级工具。
用Maven的话，核心依赖是：
<dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.3.6</version> </dependency>但在默认包名的情况下，我推荐直接javac，速度和可控性都更舒服。
4.2 准备输入数据并上传HDFS
程序的输入路径必须是HDFS路径，不是本地路径。这就是为什么前面要先做那些HDFS命令。准备一个稍大一点的输入文件更有测试价值：
echo "hadoop mapreduce spark flink hadoop hive hbase" > input.txt hdfs dfs -mkdir -p /user/demo/input hdfs dfs -put input.txt /user/demo/input可以多上传几个文件，因为MapReduce会把每个文件视为不同的输入分片，这样能看到多个MapTask并行跑的效果。
4.3 提交任务并解读运行日志
执行提交命令：
hadoop jar wordcount.jar WordCountDriver /user/demo/input /user/demo/output命令格式是固定的：hadoop jar + jar包名 + 主类名 + 业务参数。运行后你会看到一大段日志，不要被吓到，重点抓这几行：
- 包含
Job job_xxx_0001的行，这是job编号。 - 每次出现
map 0% reduce 0%到map 100% reduce 100%的进度更新。 - 结尾有
Job job_xxx_0001 completed successfully。 - 最后会打印一些计数器，比如
Map input records、Reduce output records，可以用它们核对结果是否和你预期一致。
任务成功后查看结果：
hdfs dfs -ls /user/demo/output hdfs dfs -cat /user/demo/output/part-r-00000输出目录下会出现一个part-r-00000文件，这就是reduce阶段写出的结果。还有一种情况，如果你没有设置Reducer（只做Map），结果文件会命名为part-m-00000。看到part-r前缀，说明整个流程完整跑通了。
有一点必须强调：输出路径在提交前绝对不能存在。这是MapReduce框架的保护机制，目的是防止覆盖之前的结果。
4.4 不想写Java，Python也能跑MapReduce
不少非Java背景的同学更喜欢用Python。Hadoop Streaming提供了一条方便的路径，你只需要按照约定写两个Python脚本。
mapper.py：
#!/usr/bin/env python3 import sys for line in sys.stdin: line = line.strip() words = line.split() for word in words: print(f"{word}\t1")reducer.py：
#!/usr/bin/env python3 import sys current_word = None current_count = 0 for line in sys.stdin: line = line.strip() word, count = line.split("\t", 1) count = int(count) if current_word == word: current_count += count else: if current_word: print(f"{current_word}\t{current_count}") current_word = word current_count = count if current_word: print(f"{current_word}\t{current_count}")提交命令变成：
hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \ -files mapper.py,reducer.py \ -mapper "python3 mapper.py" \ -reducer "python3 reducer.py" \ -input /user/demo/input \ -output /user/demo/output_pyStreaming适合快速验证逻辑，但如果你要精确控制MapReduce的执行细节、使用自定义类型和计数器，Java方案还是首选。我建议两条路都试一遍，理解会更立体。
5. 运行原理拆解：代码之外的关键机制
5.1 输入分片决定了MapTask的并行度
很多人以为一个文件一个MapTask。实际规则是：框架通过FileInputFormat把输入文件切成若干split，一个split生成一个MapTask。split大小默认等于HDFS块大小（128MB），也就是说，一个128MB的文件在默认配置下只生成一个MapTask，即使集群有一百个节点也只跑一个Map任务，这显然没有利用好并行度。
如果文件不大，可以通过设置mapreduce.input.fileinputformat.split.maxsize来强制切小：
conf.set("mapreduce.input.fileinputformat.split.maxsize", "67108864");这会把split切到64MB，从而让更多MapTask并行。但也不是切得越小越好，因为每个MapTask都对应一定的调度开销，split过小会导致大量小任务抢占资源，整体效率反而下降。
5.2 Map端Shuffle的两次排序和一次溢写
Map函数每写出一条记录，并不会直接通过网络传走，而是先写进一个内存环形缓冲区。缓冲区默认大小是100MB，由mapreduce.task.io.sort.mb控制。当缓冲区使用率达到80%，后台线程开始把数据溢写到本地磁盘。
溢写前会做两件事：
- 按分区号分区，目的是让相同key的数据将来能进入同一个Reducer。
- 在分区内部按键进行排序。
缓冲区会进行多次溢写，每次产生一个小文件。MapTask结束前，这些文件会被合并成一个大的溢写文件。整个Map端Shuffle复制的逻辑有点像快递网点：每个快递员收到的零散包裹先放在分拣仓库，仓库满了就打包送到中转站，最后把所有中转包裹合成一辆货车。
5.3 Reduce端Shuffle：Copy、Merge、Reduce
ReduceTask启动后会启动多个拉取线程，同时从不同MapTask取回属于自己的分区数据。数据量不大时，直接放在内存；数据量大了就落盘。之后Reduce端会把从不同MapTask取回的数据进行归并排序，最终形成一个“所有相同key连续排列”的有序迭代器，这个迭代器就是传给reduce方法的values。
这里的关键点是：Reduce方法实际能看到的数据不只是它自己的MapTask，而是所有MapTask中同一个分区的数据。所以无论你有多少个MapTask，最终结果都不会丢数据。
5.4 关键参数速查
| 参数名 | 作用 | 常见设置 |
|---|---|---|
| mapreduce.job.reduces | ReduceTask数量 | 默认1，数据量大时调到5-10 |
| mapreduce.task.io.sort.mb | Map端缓冲大小 | 默认100MB，大任务调大 |
| mapreduce.map.memory.mb | MapTask容器内存 | 默认1024MB，堆内存不足就调大 |
| mapreduce.reduce.memory.mb | ReduceTask容器内存 | 默认1024MB |
| mapreduce.input.fileinputformat.split.maxsize | 输入分片最大值 | 控制MapTask并行度 |
| mapreduce.job.map.output.collector.class | Map输出收集器 | 一般不用动 |
调参的原则永远是：先看默认值够不够，不要一上来就乱调。大多数面试和实际场景，只需要关注ReduceTask个数和内存大小。
5.5 Combiner到底能不能加，为什么
在Driver中你可以加一行：
job.setCombinerClass(WordCountReducer.class);Combiner是在Map端做一次本地Reducer，把“hello, 1”和“hello, 1”先合成“hello, 2”，减少Shuffle阶段传输的数据量。但必须满足一个前提：合并操作满足交换律和结合律，也就是局部求和的结果和全局求和的结果相等。WordCount满足这个条件。
如果业务是求平均值，就不能直接套用Reducer做Combiner。例如求平均分，局部平均和全局平均结果并不一样。这种场景要么单独写Combiner逻辑，要么放弃Combiner，老老实实把原始值全部拉过去。
6. 踩坑实录：MapReduce首跑常见的6类问题
6.1 日志里只有Exit code 1，看不到具体错误
很多新手第一次跑任务失败时都会遇到这个情况：右下角报错只给一个Container exited with a non-zero exit code: 1，没有任何栈信息。原因在于任务在NodeManager的Container里运行，本地的System.out不会直接显示在提交端。
解决方法是去查日志。先通过提交时的输出找到Application ID，然后执行：
yarn logs -applicationId application_xxx_0001syslog里会包含完整的Java异常栈，绝大多数问题在这里都能看到真相。我遇到过很多次，代码本地测试没问题，一提交就失败，最后都是在yarn日志里发现是依赖jar包缺失。
6.2 输出目录已存在的FileAlreadyExistsException
这个报错最经典。错误信息大概长这样：
org.apache.hadoop.mapred.FileAlreadyExistsException: Output directory hdfs://localhost:9000/user/demo/output already exists原因是HDFS上的输出目录已经存在。框架不会禁止你覆盖，但绝不会帮你清理。处理方式很简单，删除旧目录或换一个新路径：
hdfs dfs -rm -r /user/demo/output不要在生产环境随便删目录，最好先确认目录里的内容确实没用。
6.3 找不到主类或ClassNotFoundException
如果你在提交时写的是WordCountDriver，但jar包里根本没有这个类，会出现找不到主类。多半是编译时源文件不在当前目录，或者打jar包时class路径不对。
排查方式很简单，先用jar命令看一下包结构：
jar -tf wordcount.jar如果看到WordCountMapper.class但没有WordCountDriver.class，说明编译阶段遗漏了Driver文件，重新编译一次即可。
6.4 Java版本不匹配导致的各种诡异报错
NativeIO或者ProtocolException这类奇怪的报错，很多时候是编译版本和运行环境的JDK版本不一致。编译时用了JDK17的javac，运行时Hadoop却跑在JDK8下，class文件版本过高导致不兼容。
养成习惯，先看版本：
java -version hadoop version尽量让这两个命令的Java版本一致，不要另起炉灶装多个JDK版本互相覆盖PATH。
6.5 单词全部集中在一个Reducer，处理慢到怀疑人生
默认ReduceTask数量是1，不管MapTask有多少，最后所有数据都交给一个ReduceTask处理。如果你的输入文件很大，Map阶段很快就跑完了，但Reduce阶段卡在99%不动，大概率就是这个原因。
在Driver里显式设置ReduceTask数量：
job.setNumReduceTasks(3);WordCount这种场景可以直接设置多个ReduceTask，输出结果会变成part-r-00000、part-r-00001、part-r-00002多个文件。如果想查看总体统计，可以把所有part文件再合并一次。
6.6 常见错误速查表
| 现象 | 常见原因 | 处理方式 |
|---|---|---|
| Exit code 1，无详细堆栈 | Container内运行，日志在yarn侧 | 使用yarn logs查看syslog |
| FileAlreadyExistsException | 输出目录已存在 | 删除旧目录或换路径 |
| ClassNotFoundException | jar包没打全 | jar -tf检查，重新打包 |
| 连接被拒绝 | NameNode或ResourceManager没启动 | jps排查进程，看启动日志 |
| StackOverflow或OOM | MAP内存不足 | 调大mapreduce.map.memory.mb |
| 结果全为空 | 输入路径写错或文件为空 | hdfs dfs -ls检查输入目录 |
| 进度卡在reduce 99% | Reducer数太少或数据倾斜 | 设多个Reducer，考虑加Combiner |
最后再分享一点我自己跑WordCount的经验
我在很多次测试里发现，最快定位问题的方式不是反复盯代码，而是把提交命令、输出目录、输入数据和日志这四样东西按顺序检查一遍。很多人一遇到报错就怀疑代码逻辑，但实际上一半以上的问题集中在环境、路径和依赖上。
如果你的目标只是把MapReduce跑通，不需要追求一次性写出符合所有工程规范的代码，只需按顺序走完编译、打包、提交、看日志、查结果这套流程。等这个闭环建立起来之后，再逐步引入自定义计数器、Combiner、Partitioner、多输入格式这些进阶特性。
最后再送一个实用小技巧：调试阶段把输入文件刻意做得小一点，比如只有几行内容。这样MapTask和ReduceTask各只有一两个，日志短、跑得快，任何错误都能迅速定位。等逻辑已经确认稳定了，再换大数据量做性能验证。这是我用得最顺手的一个习惯。
