很多学大数据的人都会问我同一个问题：为什么入门教程千篇一律都是“hadoop实现单词统计”？这个需求简单到用一行脚本就能完成。我至今还记得第一次跑通这个案例时的感觉——命令行里跳出一条一条 Map 进度日志，接着 Reduce 的日志闪过，最后cat出结果文件，里面有hadoop 3、hello 2这样的输出。说实话，那个瞬间比解出一道算法题还让人开心。单词统计虽然功能朴素，但它把分布式存储、分布式计算、数据切分、任务调度、容错重试、shuffle 排序这一整套链路全部串起来了。这篇博文就是一份完整的实战记录，写给刚装好 Hadoop、还没跑通第一个 MapReduce 作业的人。

1. 单词统计为何是每个大数据人的第一道坎

1.1 需求本身：一句话能说清楚的事

单词统计的需求一句话就能讲明白：给你一批文本文件，统计每个单词总共出现多少次。比如输入三行：

hello hadoop hello hdfs hello hive world hadoop

最终输出应该是按单词字典序排列的结果：hadoop 2、hello 3、hdfs 1、hive 1、world 1。

如果用传统的 Linux 命令，这确实不算事。grep -o配合sort、uniq -c几行命令就能出结果，本地跑一个几百 MB 的日志文件也不会太慢。所以刚入门的人很容易产生一个疑问：为什么要为一个如此简单的统计引入一整套 Hadoop 体系和 Java 程序？

这个疑问本身是对的，也恰恰是 Hadoop 这个案例的价值所在。如果数据量只有几十 MB，任何方案都无所谓。可当你面对几百 GB 甚至几十 TB 的日志，文件分散在上千台机器上，传统脚本首先要解决“数据在哪”的问题。单机读取所有数据，光磁盘 IO 就会把性能拖垮。这时候需要的不再是一个统计脚本，而是一套能利用多台机器并行处理数据的框架。

1.2 它打通了分布式计算的全部关键节点

单词统计的 MapReduce 实现了看似简单、实则完整的一次数据流转：输入文件被切分到多个节点，每个节点只处理自己负责的那部分数据；处理完成后，相同单词对应的中间结果会被 shuffle 到同一个节点；最后再做一次全局聚合。整个流程中，HDFS 负责数据存储，YARN 负责资源调度和任务运行，MapReduce 负责计算模型的落地。这三个组件正是 Hadoop 生态最核心的东西。

我见过不少同学背了很多名词概念，比如分片、位压缩、心跳、容错，但一问他“Map 之后的数据去哪了”“Shuffle 到底做了什么”，就答不上来。真正亲手把一个单词统计作业跑通，这些概念就不再是干巴巴的术语了，而是你亲眼在日志里看到的过程。所以我要说：这个案例虽然简单，但它不是知识的终点，而是把抽象概念变成实感经验的必经之路。适合的人群也很明确——刚入门大数据、准备做离线计算开发、或者面试前想快速捡起 MapReduce 细节的人。

2. 本机伪分布式集群：半小时搭出一张可运行的作业流水线

2.1 版本选择：别在这里浪费太多时间

联网看教程最怕的就是版本混乱。Hadoop 1.x 已经彻底淘汰，我们不做任何考虑。2.x 目前在生产环境还有很多存量，但新学习者建议直接用 Hadoop 3.x，我实际使用的是 3.3.6。JDK 方面使用 JDK 8，这是 Hadoop 3.x 兼容最稳妥的选择。Java 版本不是越高越好，比如 JDK 17 在某些模块上会和 Hadoop 存在兼容问题，你折腾半天可能不是自己的问题，而是版本组合问题。按照“JDK 8 + Hadoop 3.3.x”这个组合来，基本不会踩坑。

硬件方面，伪分布式模式只需要一台 Linux 机器，可以是虚拟机，也可以是云服务器。内存建议至少 8 GB，因为这个模式会在本机同时启动 NameNode、DataNode、ResourceManager、NodeManager 等多个 JVM 进程，内存太小会出现容器被频繁杀掉的诡异问题。操作系统我用的是 CentOS 7.9，Ubuntu 的配置过程也类似，只是包管理器命令不同。

2.2 环境变量与四个配置文件的关键设置

先得把 Hadoop 解压到一个固定目录，比如/opt/hadoop。然后编辑/etc/profile或用户目录下的~/.bashrc，加上：

export JAVA_HOME=/usr/local/jdk1.8 export HADOOP_HOME=/opt/hadoop export PATH=$PATH:$JAVA_HOME/bin:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

这里最容易忘的是把sbin目录也加进 PATH，否则后面执行start-dfs.sh会提示找不到命令。另外hadoop-env.sh文件里也需要显式指定JAVA_HOME，因为 Hadoop 的启动脚本不会自动读取系统变量。很多人在start-dfs.sh阶段报错找不到 JDK，基本都是这个原因。

接着修改etc/hadoop目录下的四个核心配置文件。core-site.xml是全局配置，要指定默认文件系统地址：

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

hdfs-site.xml是 HDFS 相关配置。伪分布式模式下副本数设置为 1，否则一个 DataNode 节点存不下三份副本会导致各种告警：

<configuration> <property> <name>dfs.replication</name> <value>1</value> </property> <property> <name>dfs.namenode.name.dir</name> <value>/opt/hadoop/data/name</value> </property> <property> <name>dfs.datanode.data.dir</name> <value>/opt/hadoop/data/data</value> </property> </configuration>

yarn-site.xml是资源调度配置，最主要是把yarn.nodemanager.aux-services设置为mapreduce_shuffle，否则 MapReduce 作业无法运行：

<configuration> <property> <name>yarn.nodemanager.aux-services</name> <value>mapreduce_shuffle</value> </property> </configuration>

mapred-site.xml是 MapReduce 的运行框架配置，让作业运行在 YARN 上：

<configuration> <property> <name>mapreduce.framework.name</name> <value>yarn</value> </property> </configuration>

2.3 初始化、启动和验证

首次启动前需要配置本机 SSH 免密登录。Hadoop 启动脚本会通过 SSH 登录到各个节点启动守护进程，即使是在单机伪分布式模式下也需要这个前提。在终端执行：

ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys chmod 600 ~/.ssh/authorized_keys

然后做格式化。注意：只能在首次启动前格式化一次，之后如果反复执行hdfs namenode -format，NameNode 的命名空间 ID 会变化，DataNode 无法对应上，就会出现连不上 NameNode 之类的诡异错误。

hdfs namenode -format

格式化完成后启动服务：

start-dfs.sh start-yarn.sh

输入jps命令验证进程，正常情况下能看到五个关键进程：NameNode、DataNode、SecondaryNameNode、ResourceManager、NodeManager。再打开浏览器访问 NameNode 的 Web 界面（Hadoop 3.x 默认端口是 9870）和 ResourceManager 界面（默认 8088）。页面能打开说明 HDFS 和 YARN 已经就绪，接下来就可以开始跑作业了。

3. 核心机制拆解：一条文本如何变成键值对并完成聚合

3.1 Map 阶段：从行文本到最小的计数单位

很多人看 WordCount 的代码觉得太简单，无非是context.write(word, 1)。但这一步为什么这么设计，值得想明白。

在 Map 阶段，输入文件会被框架切成若干个 split 分片。默认情况下，一个 split 对应一个 map task。每个 mapper 读取分片时，TextInputFormat会按行读取，把每一行封装成一个键值对：key 是该行在文件中的字节偏移量（LongWritable），value 是这一行的完整文本（Text）。也就是说，Map 阶段的输入天然就是键值对的流式处理。

WordCount 的 Mapper 拿到一行文本之后，用分词器把文本切成一个个单词，每遇到一个单词就输出一次(word, 1)。这是 MapReduce 的通用设计哲学：你在 Map 阶段只输出中间结果，不负责最终累加。因为同一个单词很有可能出现在不同机器处理的多个分片中，对自己那部分只做了局部分析，需要把结果送到同一个地方做全局汇总。输出(word, 1)这种可累加的最小单位，正是为了让后续的聚合变得简单可靠。

举一个生活例子：一个班有 100 个学生要统计全班每月读书总数。最笨的方式是每位同学跑到讲台上大声报出自己的阅读量，班长听到后逐个加起来。如果所有同学同时冲到讲台上，班长一个人就处理不过来。MapReduce 的 Map 阶段相当于每个宿舍先统计自己宿舍的数据，Reduce 阶段相当于汇总各宿舍的结果。为了让汇总不出错，大家要先约定一种统一的提交格式：每次提交一个“本月读了几本书”的数字，最后逐项相加。

3.2 Shuffle 阶段：框架里最贵的那段传输

Map 阶段输出的中间键值对不会直接送到 Reduce 阶段，中间隔着极其重要的 Shuffle 过程。这个过程中要依次完成分区、排序、溢写、合并、拉取、归并等操作。

先看分区。Map 输出的记录会根据 key 的哈希值决定进入哪个 Reduce 分区，默认的HashPartitioner逻辑就是key.hashCode() % reduceTaskNum。分区意味着同一个单词的所有中间记录会被分到同一个分区，最终进入同一个 Reducer，这样才能完成全局累加。如果没有这一步，不同机器上的同一单词被送到不同 Reducer，那最终结果就乱了。

再来看排序。Map 端输出首先会写入一个缓冲区，默认大小约 100 MB。当缓冲区使用率达到 80% 时，就会触发溢写，把缓冲区数据按分区、按 key 排序后写到磁盘。排序的好处是不同 map task 产出的数据到达 Reduce 端后，可以直接做归并排序，不用再做代价更高的全量排序。整个过程在 MapReduce 运行日志中表现为map 100% reduce 66%，Reduce 的进度会停顿在 66%，因为它在等待和拉取 map 输出，这也被称为 Shuffle 的“Copy 阶段”。

可以这样理解 Shuffle：一群外卖小哥（不同 map task）把订单送到一个总仓库（Reduce 节点），仓库接收员收到订单后不是乱糟糟地堆放，而是按菜品名称排好。同一个菜品的订单能够放在相邻的位置，后续统计同一个菜品销量时，只需要顺着同一堆订单相加即可。

3.3 Reduce 阶段：把中间结果变成最终答案

Reduce 阶段从每个分区中获取属于自己那一部分的数据，经过再次合并排序后，框架会把相同 key 的所有 value 分到一组，调用一次 reduce 函数。WordCount 的 Reducer 做的事情很简单：遍历迭代器里的所有 value，把它们累加起来，输出(word, sum)。

这里有个细节需要留意：Map 端输出的 key 是Text，Reducers 收到相同单词的所有 value 时，value 的类型是IntWritable。为什么 Hadoop 不使用 Java 原生的String和Integer？因为 Hadoop 需要跨节点传输数据，对象必须能够序列化，而 Java 原生对象的序列化机制体积大、性能差。Hadoop 自定义了一套序列化框架（Writable），例如Text对应字符串，IntWritable对应整数，它们的序列化结果紧凑且解析速度快。所以你在 Mapper 和 Reducer 的泛型签名里看到的都是这种类型，而不是常见的 String、Integer。如果你写过网络协议或者 RPC 调用，应该对这种空间换性能、定制序列化的思路很熟悉。

最终 Reducer 的输出会写到 HDFS 上。有多少个 Reducer 就会生成多少个part-r-xxxxx文件。我们最常见的单词统计作业若只设置了 1 个 Reducer，最终只有一个输出文件。这个文件的下游使用者只需要按行读取，格式是“单词 + 制表符 + 数量”。

4. 代码实现：WordCount 的三个类到底该怎么写

4.1 Maven 依赖：只用 provided scope

不管是用 Maven 还是 Gradle，最重要的依赖是hadoop-client。我用 Maven 时依赖声明如下：

<project> <modelVersion>4.0.0</modelVersion> <groupId>com.example</groupId> <artifactId>wordcount</artifactId> <version>1.0-SNAPSHOT</version> <dependencies> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.3.6</version> <scope>provided</scope> </dependency> </dependencies> </project>

注意<scope>provided</scope>这一行。Hadoop 集群环境本身已经有客户端库，打包时不需要把这些依赖打进去。如果不想用 provided，也可以打一个包含所有依赖的 fat jar，但对当前案例来说完全没有必要。

4.2 Mapper 实现：输出的 key 和 value 类型不能想当然

WordCount 的 Mapper 要继承org.apache.hadoop.mapreduce.Mapper，新版 API 的 Mapper 类有四个泛型参数：输入 key 类型、输入 value 类型、输出 key 类型、输出 value 类型。对应到单词统计就是：

public class WordCountMapper 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 { String line = value.toString(); StringTokenizer itr = new StringTokenizer(line); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } }

LongWritable作为输入 key 类型，表示行偏移量；Text是行内容。输出 key 选择Text，输出 value 选择IntWritable。这里特别提醒一点：word和one这两个对象是在类的成员变量里定义的，map 方法内部只是不断重用这两个对象，而不是每次循环都new。这是 Hadoop 编程里一个很重要的性能习惯，因为 map 方法会被调用无数次，如果每次都创建新对象，JVM 的垃圾回收压力会很大。

StringTokenizer默认按空格、制表符、换行符等空白字符切分单词，足够满足英文单词统计的场景。如果输入是中文文本，你会发现整个句子被当成一个 token，那就需要引入分词器了，这个后面章节会详述。

4.3 Reducer 实现：同一单词的所有 value 一起进来

Reducer 的泛型参数是输入 key、输入 value、输出 key、输出 value。输入 key 就是单词Text，输入 value 是一串数字IntWritable，输出 key 同样是单词，输出 value 是累加结果。代码：

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); } }

这个类可以完全当作 Combiner 使用，因为“求和”运算是满足交换律和结合律的。Combiner 的作用是 Map 端本地聚合，比如某个 mapper 里连续输出了 100 次hello，如果没有 Combiner，这 100 个(hello, 1)都要通过网络传输；有了 Combiner 后，map 端会先就地累加成(hello, 100)，再送给 Reducer。生产环境中这种优化对降低网络 IO 非常明显。但要注意：Combiner 并不是所有场景都能直接复用 Reducer，比如求平均值的场景，如果用 Reducer 当 Combiner，局部每个 Map 的平均值合起来并不是全局平均值。WordCount 里累加场景没问题，但如果你把它当通用规律去套，迟早会踩坑。

4.4 作业驱动类：Job 配置的六个必填项

最后一个类是 Main 函数，负责组装作业配置。对于刚入门的同学来说，这个类比 Mapper 和 Reducer 更容易报错，因为太多配置项依赖运行时环境，排错需要经验。

public class WordCountDriver { public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordCountDriver.class); job.setMapperClass(WordCountMapper.class); job.setCombinerClass(WordCountReducer.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); } }

这里我要讲一下setOutputKeyClass和setOutputValueClass为什么必须写。如果你只设置了 Reducer 的输出类型，Mapper 的输出类型没有单独指定，框架默认认为 Mapper 输出类型和 Reducer 输出类型一致。而 WordCount 里 Mapper 和 Reducer 的输出恰好是一致的，所以这两个设置可以同时生效。如果你某个作业的 Mapper 和 Reducer 输出类型不一样，还得额外调用job.setMapOutputKeyClass和job.setMapOutputValueClass单独指定，否则运行时会报类型不匹配的错误。

输入输出路径是从args命令行参数里读取的，所以运行命令时必须在 jar 包后面跟上两个路径参数：输入目录、输出目录。如果输出目录已经存在，作业会直接报错，这是有意设计的，为了防止误覆盖之前的结果。

5. 从打包到看见结果：一次完整调试链

5.1 HDFS 准备：先让输入文件待在该待的地方

写好了代码，在 IDE 里跑通本地模式还不够。既然搭建了分布式模式的集群，就要严格按照分布式作业的流程走：数据先上传到 HDFS，然后作业从 HDFS 读取，结果再写回 HDFS。

在项目根目录执行打包：

mvn clean package -DskipTests

随后在 target 目录会生成wordcount-1.0-SNAPSHOT.jar。接下来准备输入数据。我先建了一个本地文件test.txt，内容包含几种单词的混合：

echo -e "hello hadoop\nhello hdfs hello hive\nworld hadoop" > test.txt

然后创建 HDFS 目录并上传：

hdfs dfs -mkdir -p /wc/input hdfs dfs -put test.txt /wc/input hdfs dfs -ls /wc/input

如果ls能看到 test.txt，说明 HDFS 写入正常。一个小经验：在提交作业之前用hdfs dfs -cat /wc/input/test.txt确认文件内容符合预期，别等作业跑完才发现输入文件是空的或者乱码。这种检查成本极低，收益却很大。

5.2 提交作业：日志里藏着全流程的信息

运行作业的命令：

hadoop jar target/wordcount-1.0-SNAPSHOT.jar com.example.wordcount.WordCountDriver /wc/input /wc/output

第一次跑的时候，推荐不要设置安静模式，让控制台把所有输出打出来。你会看到类似下面的关键信息：

INFO mapreduce.Job: Running job: job_1710000000000_0001 INFO mapreduce.Job: map 0% reduce 0% INFO mapreduce.Job: map 100% reduce 0% INFO mapreduce.Job: map 100% reduce 100% INFO mapreduce.Job: Job job_1710000000000_0001 completed successfully

这个过程中，如果你只开启了 1 个 Reducer，Reduce 阶段会卡在 66% 较长时间，那不是程序卡死，而是在执行 Shuffle 拉取数据。当年我第一次看到 reduce 66% 还以为作业失败了，其实是正常状态。

5.3 验证输出：正确结果只是第一步

作业跑完后，输出目录下会生成结果文件：

hdfs dfs -ls /wc/output hdfs dfs -cat /wc/output/part-r-00000

预期输出结果应该是单词按字典序排列：

hadoop 2 hdfs 1 hello 3 hive 1 world 1

如果结果和你设想的不一致，最快的排查方式是检查输入文件里有没有多余的空格、大小写差异、标点符号粘连。比如 “hello,” 和 “hello” 会被StringTokenizer当成两个不同的单词，因为它们确实不是同一个字符串。这个细节也引出一个经典改进方向：Map 端先做一次单词清洗，把标点符号去掉、统一转小写，统计结果才会符合业务预期。

5.4 常见运行错误：每一张都是血泪史

我整理一份常见错误对照表，看看你能不能提前避开：

这一类错误经验非常值得积累。我见过某同学在反复格式化 NameNode 后，DataNode 始终无法注册，最后只能清空数据目录重来；也见过输出目录一旦存在就不知所措的。其实 Hadoop 的报错信息已经很友好，关键是沉住气从上往下读日志，而不是一上来就怀疑环境有问题。

6. 别停在 Demo：词频统计的进阶优化方向

6.1 让 Shuffle 更轻：Combiner 和中间结果压缩

单词统计在数据量很小的验证阶段，Combiner 没有存在感。但当你把输入换成几百 GB 的日志，Map 端每个节点都会产生海量的(word, 1)，把这些中间键值对全部通过网络传输到 Reduce 端，代价不可忽略。设置了 Combiner 之后，每个 Map 节点内部先做一轮本地聚合，网络传输的数据量往往能减少一个数量级。

中间结果也可以开启压缩。和输出文件的压缩不同，Map 和 Reduce 之间的数据属于临时数据，压缩后虽然增加了少量 CPU 开销，但降低了磁盘和网络的消耗。在实际大作业中，这个收益通常大于开销。可以在mapred-site.xml中配置：

<property> <name>mapreduce.map.output.compress</name> <value>true</value> </property> <property> <name>mapreduce.map.output.compress.codec</name> <value>org.apache.hadoop.io.compress.SnappyCodec</value> </property>

Snappy 压缩速度极快，适合 Map 端输出这类场景。

6.2 处理中文文本：真正的 WordCount 业务往往要先分词

很多实际场景下，用户给的并不是英文语料，而是中文的文章、搜索词、评论。若直接套用StringTokenizer，"我爱北京"整句会被当成一个“单词”，这显然不是业务想要的“词频统计”。正确的处理链条是：中文分词 + MapReduce 统计。

中文分词可以用开源分词工具，在 Map 阶段对行文本分词后，每个词输出(词, 1)。这并不需要高级技巧，只需要把 Mapper 里的StringTokenizer替换为分词接口，返回一个词列表，然后循环输出即可。这里我多说一句：很多人在这个环节容易忽略停用词过滤，像“的”“了”“在”这种高频无意义词会霸占统计结果前列，需要在分词后先过滤再输出。

6.3 改一版更容易读懂的实现：Hadoop Streaming

如果你对 Java 实在不熟悉，Hadoop 也提供了 Streaming 方式，允许用 Python、Shell 等语言编写 Mapper 和 Reducer。它的核心理念是：从标准输入逐行读取数据，处理之后写入标准输出，框架再去拼接这些数据流。

Python 版的 mapper 可以像这样写：

for line in sys.stdin: for word in line.split(): print(f"{word}\t1")

reducer 则是按单词累计：

current_word = None current_count = 0 for line in sys.stdin: word, count = line.strip().split("\t", 1) if current_word != word: if current_word: print(f"{current_word}\t{current_count}") current_word = word current_count = int(count) else: current_count += int(count) if current_word: print(f"{current_word}\t{current_count}")

运行命令：

hadoop jar /opt/hadoop/share/hadoop/tools/lib/hadoop-streaming-3.3.6.jar \ -mapper mapper.py \ -reducer reducer.py \ -input /wc/input \ -output /wc/stream_output

Streaming 的坑在于不能直接执行本地的.py文件位置，需要先上传到 HDFS 或者使用-file参数分发脚本。这一点如果没搞清楚，运行时会报找不到文件的错误。

6.4 按词频排序：一次统计的常见扩展需求

单词统计输出是按单词字母序排列的，这在词频分析场景往往不够。你可能更想知道哪些词出现次数最多，这就要按 value 排序。最简单的方案是使用第二个 MapReduce 作业：第一个作业输出(word, count)，第二个作业把 count 作为 key、word 作为 value，利用 Hadoop 的排序机制实现按 count 降序。此时需要自定义一个IntWritable比较器，因为默认的整数排序是升序。

如果嫌写第二个作业麻烦，小数据量也可以把part-r-00000拉回本地处理：

hdfs dfs -cat /wc/output/part-r-00000 | sort -t $'\t' -k2 -nr | head -20

这种本地解法只能作为临时工具，真正生产环境还是得落在分布式处理框架里。

我在实际练习中体会最深的一点是：跑通 WordCount 不算什么本事，跑通之后愿意从 Combiner、压缩、小文件、分词、排序这些角度重新审视它，才算真正开始理解分布式计算。如果你现在能独立把这个案例的所有细节讲给别人听，并且能说出某一步为什么这么做，那你的第一课才算真正毕业了。下一步可以尝试用 Spark 实现同一个需求，把 HDFS 上已经准备好的文件直接读进去，用RDD的flatMap和reduceByKey完成统计。两个方案对比着看，你会对 MapReduce 的优势和限制都有更具体的认识。