简介一套基于Hadoop 2.2.0的完整词频统计MapReduce解决方案面向正在入门分布式计算、希望掌握MapReduce编程流程的开发者。该方案覆盖从原始文本读取到最终结果输出的完整链路能够解决在集群环境下统计单词出现次数的经典问题是理解分布式数据处理逻辑的极佳范例源码与编译后的字节码文件同时提供方便逐行对照。压缩包内共包含十七个文件具体包括七份Java源程序文件、七份编译后生成的字节码文件、一份约含十万个单词的英文测试文本以及Eclipse工程所需项目配置与类路径说明文件目录划分清晰可直接导入集成开发环境运行验证。整个压缩包仅154KB体量轻巧内容聚焦特别适合课程实验或面试前的快速复习目前已有五千八百六十五人学习下载说明该资源受到较多学习者关注。借助内置测试文本读者可以直观观察映射器、归约器与组合器等核心组件如何协同完成计数任务并通过替换测试数据、修改键值类型等方式进一步加深理解从入门走向熟练运用。 但凡用Hadoop的人第一个跑通的MapReduce程序几乎都是词频统计WordCount。它看起来就是个“数单词”的小任务但背后把HDFS、MapReduce、YARN这三大组件全部串了起来所以不管是课程设计、面试准备还是入门Hadoop开发这个词频统计都是绕不开的起点。我这篇文章就想给你一份能直接照着跑的完整流程从环境准备、HDFS数据操作、Java代码实现到编译打包、提交运行、结果查看再到我实际踩过的各种坑全部一条龙说清楚。适合刚搭完Hadoop环境、准备跑第一个作业的人也适合那些“代码能抄但跑不通”的同学用来排查问题。1. 先理清任务整体思路WordCount到底在做什么1.1 一个词频统计任务拆开看词频统计的目标很简单给一批文本统计每个单词出现的次数。比如输入“hello world hello hadoop”输出结果应该是“hadoop 1hello 2world 1”。在单机上这用几行循环就能写完但到了大数据场景数据量可能达到TB甚至PB级单机内存根本放不下必须把数据切成很多份分给多台机器并行处理最后再汇总。MapReduce就是干这件事的框架它的核心思想是“先分后合”Map阶段负责拆分和映射Reduce阶段负责汇总。这个词频统计任务恰好就是MapReduce最经典的“Hello World”。为什么经典因为它的逻辑足够简单方便你把注意力放在框架本身而不是业务算法上。你把WordCount跑通了就等于理解了MapReduce的运行机制数据怎么从HDFS被读进来、怎么被切分成键值对、中间结果怎么排序和分组、最终怎么写成文件。这些理解了后面再写更复杂的MapReduce作业就有底了。1.2 方案选型为什么用Java写怎么跑最合适我身边不少人问过词频统计用Python写行不行当然行Hadoop Streaming可以让你用Python、Shell等语言写Mapper和Reducer。但如果你是想深入理解MapReduce的原理我建议第一版用Java写。原因很直接Hadoop本身就是Java写的Java API的示例最齐全、官方文档最完整你搜问题的时候也最容易搜到同类案例。另外Java版本能让你直接看到Mapper、Reducer、Job这几个核心类是怎么协作的逻辑最清晰。运行模式上我建议区分两种情况。如果只是做功能验证、跑小数据量测试可以用本地模式不需要启动HDFS和YARN直接在本地文件系统上跑速度快、调试方便。如果是为了模拟真实环境、做大作业或面试准备就要用伪分布式模式在一台机器上启动所有Hadoop守护进程数据放HDFS作业提交给YARN调度。伪分布式和真实集群在开发层面几乎没区别你写完的代码直接拿到集群上也能跑所以这篇我会把两种模式都讲到。2. 环境准备从基础检查到HDFS数据落盘2.1 Hadoop基础环境检查清单动手写代码之前先把基础环境确认一遍不然代码编译过了也跑不起来。我在新机器上通常会依次执行下面几条命令java -version echo $JAVA_HOME hadoop version jps要注意几点Java版本建议用Java 8或Java 11Hadoop 3.x对这两个版本支持最稳我遇到过有人用Java 17跑Hadoop 3.1结果连NameNode都起不来。JAVA_HOME必须显式配置好Hadoop启动脚本很多地方会直接引用这个变量。jps这条命令很关键它能看到当前机器的Java进程如果是在伪分布式模式下应当能看到NameNode、DataNode、ResourceManager、NodeManager、SecondaryNameNode这几个进程缺了哪个说明对应组件没起来。如果你还没启动Hadoop伪分布式模式下需要执行start-dfs.sh start-yarn.sh跑完再用jps确认进程然后可以用hdfs dfsadmin -report看看DataNode状态是否正常。这一步很多人会漏直接把文件往HDFS上传结果报错说Connection refused其实就是NameNode没启动。2.2 HDFS数据准备创建目录和上传文件Hadoop作业的输入数据一般要放到HDFS里这样数据才是“分布式存储”的。先用命令创建一个测试文件内容随意英文文本就好注意单词之间用空格或换行分隔echo hello world hello hadoop hadoop mapreduce hdfs yarn hello wc_input.txt然后上传到HDFShdfs dfs -mkdir -p /input hdfs dfs -put wc_input.txt /input/ hdfs dfs -ls /input/ hdfs dfs -cat /input/wc_input.txt-mkdir -p /input是递归创建目录如果目录已存在也不报错。-put是上传-cat是查看文件内容这三步连起来能确认你的HDFS读写正常。如果你这一步遇到Permission denied说明当前用户在HDFS上没有写权限常见的临时解决方案是启动服务时要保证当前用户对HDFS根路径有权限或者用hdfs dfs -chmod -R 777 /放开权限但生产环境别这么干本地测试图省事倒没问题。2.3 输出路径规划的注意事项新手最容易忽略的一点就是MapReduce的输出目录不能提前存在。框架要求输出路径必须是“全新的”否则直接报FileAlreadyExistsException它这么设计是为了防止覆盖掉上一次的结果。所以我每次跑作业之前如果上一次已经跑过一遍会先执行hdfs dfs -rm -r /output把这个习惯刻进肌肉记忆里能帮你省下不少排查时间。另外我习惯把输入目录和输出目录分开命名比如/input和/output简单直观多个作业并存的时候也不容易搞混。等你作业跑得多起来了你会发现路径管理看似不起眼但一旦路径写错排查起来比代码报错还费劲因为有些错误信息藏得比较深。3. 核心代码实现Mapper、Reducer、Driver逐行解析3.1 Mapper阶段把文本拆成键值对写WordCount的Java代码最核心的就是三个类Mapper、Reducer以及负责组装作业的Driver也叫Main类。我们先把完整代码贴出来再逐段解释。import java.io.IOException; import java.util.StringTokenizer; 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 WordCount { public static class TokenizerMapper extends MapperObject, 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); } } } public static class IntSumReducer extends ReducerText, IntWritable, Text, IntWritable { private IntWritable result new IntWritable(); public void reduce(Text key, IterableIntWritable values, Context context ) throws IOException, InterruptedException { int sum 0; for (IntWritable val : values) { sum val.get(); } result.set(sum); context.write(key, result); } } 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); } }先看Mapper部分。MapperObject, Text, Text, IntWritable有四个泛型参数分别是输入键类型、输入值类型、输出键类型、输出值类型。输入键是行偏移量一般用Object或者LongWritable都行输入值是一行文本类型是Text。输出键是一个单词输出值是固定数字1所以类型是Text和IntWritable。map方法做的事情很简单把一行文本value转成String用StringTokenizer按空格、制表符、换行等分隔符拆成一个个单词每拆出一个单词就写一次context.write(word, one)相当于给这个单词计了个1。这里有个细节word变量是复用同一个Text对象而不是在循环里new新对象这样做是为了减少对象创建开销在大数据量下这个优化能明显降低GC压力算是MapReduce代码里的常见写法。3.2 Reducer阶段把统计结果汇总再来看Reducer部分。ReducerText, IntWritable, Text, IntWritable的四个泛型很好理解输入键值对是Map阶段输出的单词和1输出键值对是单词和累加后的总数。reduce方法接收的values是一个IterableIntWritable里面装着同一个单词对应的所有1。比如“hello”这个词在Map阶段可能被写出来3次那进入Reducer的时候values就是[1, 1, 1]我们把它们加起来得到3然后输出hello 3。你可能会问同一个单词是怎么被汇总到一起的这是MapReduce框架自动完成的工作Map阶段输出的键值对会经过一个叫Shuffle的过程框架会把相同key的value归到同一个Reducer里这个过程对开发者是透明的。理解了这一点你会发现Reducer写得再简单不过本质上就是“对同一key下的所有value做聚合”。词频统计是求和聚合其他任务可能是求最大值、去重、排序套路一样换一下聚合逻辑就行。3.3 Driver与Combiner优化作业装配的艺术main方法就是Driver负责把整个作业装配起来。要理解这部分的逻辑你得知道一个MapReduce作业的完整生命周期Job对象建立后框架会根据输入路径去HDFS上读取数据把数据切分成一个个split分配给多个Map任务处理Map输出的中间结果经过排序、分组后交给Reduce任务。所以Driver里最关键的是把Mapper类、Reducer类、输入输出类型都告诉框架。job.setCombinerClass(IntSumReducer.class)这行很多人会忽略但它对作业性能影响很大。Combiner可以理解成“本地Reducer”它运行在Map任务结束后、数据传输到Reducer之前的这个阶段。作用是在Map端先做一次局部合并比如某个Map处理的一小份数据里“hello”出现了3次Combiner会先把这3个1合并成3再传输出去这样跨节点传输的数据量就减少了。在我们这个任务里Combiner的逻辑和Reducer完全一样所以可以直接复用IntSumReducer。但要注意不是所有任务都能这样复用只有聚合操作满足交换律和结合律的时候才可以比如求和、求最大值但如果你的Reducer逻辑是求中位数那就不能随便加Combiner。4. 编译打包与运行调试从jar包到跑完整个作业4.1 Maven工程配置与打包Java代码写好了接下来要把它编译打包成jar包。我强烈建议用Maven管理项目因为在真实开发里你大概率还会引入其他依赖Maven把整个过程标准化了不会像手工javac那样后期维护成本高而且遇到类冲突、缺依赖这些问题时Maven的依赖管理机制能帮你理清楚很多麻烦。基础目录结构是这样的wc/ ├── pom.xml └── src/main/java/WordCount.javapom.xml里只需要配置两个核心依赖和编译插件我贴一个最小可用的版本hadoop-client的作用是把HDFS、MapReduce、YARN相关客户端依赖全部引入省得一个个找project xmlnshttp://maven.apache.org/POM/4.0.0 xmlns:xsihttp://www.w3.org/2001/XMLSchema-instance xsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd modelVersion4.0.0/modelVersion groupIdcom.example/groupId artifactIdwordcount/artifactId version1.0/version packagingjar/packaging properties maven.compiler.source8/maven.compiler.source maven.compiler.target8/maven.compiler.target project.build.sourceEncodingUTF-8/project.build.sourceEncoding /properties dependencies dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.3.6/version scopeprovided/scope /dependency /dependencies build plugins plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-compiler-plugin/artifactId version3.8.1/version /plugin /plugins /build /project注意到hadoop-client的scope是provided意思是编译时需要但打包时不需要打进去因为Hadoop运行环境里已经有了。如果你误把它打进了jar反而可能因为包冲突导致运行时奇怪的问题。打包执行mvn clean package成功后target/目录下会出现wordcount-1.0.jar。用ls -l target/*.jar确认一下文件在别急着下一步先记下这个jar包的路径。4.2 先在本地模式跑一遍拿到jar包后我建议先不要直接提交到Hadoop集群而是先在本地模式快速验证代码逻辑。本地模式不需要启动HDFS和YARN输入输出都在本地文件系统上运行速度极快调试问题是最方便的。准备一个本地输入文件echo hello world hadoop mapreduce hello ./input.txt然后运行注意直接把输入输出路径写成本地路径hadoop jar target/wordcount-1.0.jar WordCount ./input.txt ./output_local如果一切正常你会看到日志滚动最后出现类似Job complete的信息然后输出目录里会出现一个part-r-00000文件cat ./output_local/part-r-00000本地模式的作用不只是“先看一眼结果”它还能帮你区分“代码逻辑问题”和“分布式环境问题”。如果你在本地模式跑通了那代码本身OK后面如果伪分布式跑失败问题大概率出在HDFS路径、权限或YARN配置上这个思路能显著缩短排查范围。4.3 提交到HDFS与YARN的完整过程本地模式验证通过之后我们把输入文件放到HDFS上然后正式提交到Hadoop。假设前面已经把wc_input.txt上传到/input此时执行hadoop jar target/wordcount-1.0.jar WordCount /input /output注意这里输入输出路径以/开头表示HDFS的根路径。如果你在本地模式已经生成了./output_local不用担心那是本地文件系统的路径和HDFS路径互不干扰。运行过程中你可以看到进度日志大概是这样INFO mapreduce.Job: Running job: job_1700000000000_0001 INFO mapreduce.Job: map 0% reduce 0% INFO mapreduce.Job: map 100% reduce 100% INFO mapreduce.Job: Job job_1700000000000_0001 completed successfully看到completed successfully就说明作业跑通了。查看结果hdfs dfs -cat /output/part-r-00000我执行完上面这一步输出是这样hadoop 2 hello 3 hdfs 1 mapreduce 1 world 1 yarn 1说明数据从HDFS读取、经过MapReduce计算、结果写回HDFS整个链路是通的。到这一步你就已经完整跑通了一次Hadoop词频统计。如果你在运行日志里看到了YARN的Web UI地址默认是http://localhost:8088也可以打开浏览器看看作业的历史记录、Map数量、Reduce数量、运行耗时这些信息。这虽然是附加操作但对理解作业调度很有帮助。5. 常见问题速查与排错实录5.1 错误速查表我把自己踩过的坑和身边同学朋友遇到的高频问题整理成了表格遇到问题了直接对着查。错误信息原因分析解决方案FileAlreadyExistsException: Output directory ... already exists输出目录已存在框架默认拒绝覆盖hdfs dfs -rm -r /output删除已有目录或换一个新路径java.net.ConnectException: Connection refusedNameNode或ResourceManager没启动或者端口不对执行jps检查进程启动start-dfs.sh和start-yarn.sh确认core-site.xml里的端口配置Permission denied当前用户对HDFS路径没有写权限本地测试可执行hdfs dfs -chmod -R 777 /或把dfs.permissions设为false生产环境不建议There is no such file or directory: .../mapred/tmp/hadoop-yarn或相关系统目录权限不对在HDFS上创建目录并授权比如hdfs dfs -mkdir -p /user、/tmp再给当前用户赋权ClassNotFoundException运行时指定了错误的类名或jar里没有包含对应类检查类名是否带完整包路径用jar tf查看jar包类列表RemoteException: java.io.FileNotFoundException输入路径不存在先执行hdfs dfs -ls /input确认路径存在再检查路径拼写作业卡在map 100% reduce 0%不动数据倾斜、Reducer启动慢或节点内存不足数据量小的时候可以检查日志数据量大时考虑增加Reducer数量或调优参数上面这个表格里我自己遇到最多的是前两项尤其是FileAlreadyExistsException基本上每次调试重跑都会遇到后来我直接会在运行命令前先把输出目录删掉形成一个固定习惯。另外一个常被忽略的是类名问题如果你在代码里声明了package com.example那运行命令里的类名要写全限定名com.example.WordCount很多新手在这里栽过跟头。5.2 实际操作中的几条经验最后分享几条我在实际操作中总结的经验算不上什么高深技巧但确实能减少很多折腾。第一小数据量调试时优先用本地模式跑因为启动HDFS和YARN本身就要花不少时间频繁在分布式模式下调试会非常痛苦本地模式几秒钟就能出结果代码逻辑一清二楚。第二记得看日志文件和Web UI作业跑失败时YARN会把错误日志写到logs/userlogs目录下那里有真正的Map和Reduce日志比只看控制台信息有用得多很多异常信息控制台只给个概要细节都在日志里。第三跑大作业前先小数据量验证我习惯先用几十KB的数据确认流程没问题再换全量数据跑否则一个几小时的作业跑到一半挂了排查起来代价太高。第四jps这个命令几乎是你检查环境的启动器进程不齐全后面所有操作都白搭跑之前先看一眼比报错后猜问题效率高太多。把词频统计完整跑通之后你会发现后面写MapReduce作业都像套模板一样Mapper负责拆分、Reducer负责聚合、Driver负责装配无非是把业务逻辑换一换。这套流程跑通一次后面不管是课程设计还是工作需要都省心很多。本文还有配套的精品资源点击获取
