刚接触Hadoop的人看到“wordMapper”“wordReduce”这种叫法其实对应的就是MapReduce里最经典的WordCount程序统计一堆文本中每个单词出现了多少次。它看起来简单但如果你能亲手把这段代码写完、编译、打包、提交到集群上跑通MapReduce的核心机制基本就打通了一半。这篇文章我按照平时真实跑任务的顺序来写从Mapper和Reducer怎么设计与实现到用什么命令提交、怎么读日志、踩了哪些坑一次讲透。适合第一次接触Hadoop、想把MapReduce程序真正跑起来的读者。1. 先想清楚WordCount的“MapReduce”到底在拆解什么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时代用JDK8Hadoop 3.x也支持JDK8但部分新功能需要JDK11。我的建议是选择一个稳定组合Hadoop 3.3.x 配 JDK8 或 JDK11这是目前最常用、遇到问题最容易搜到解决方案的组合不要一味追新。推荐使用Hadoop 3.3.6这类成熟版本。太老的2.7系列虽然教程多但和现代操作系统的兼容性越来越差太新的版本可能和某些生态组件还没适配。稳定压倒一切。2.3 解压、配置环境变量、准备启动脚本第一步下载Hadoop压缩包解压到某个目录假设是/opt/hadooptar -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 namefs.defaultFS/name valuehdfs://localhost:9000/value /property /configurationhdfs-site.xml里设置副本数伪分布式下副本数为1就够了configuration property namedfs.replication/name value1/value /property /configurationyarn-site.xml里启用YARN资源调度并设置NodeManager的辅助服务configuration property nameyarn.nodemanager.aux-services/name valuemapreduce_shuffle/value /property property nameyarn.nodemanager.aux-services.mapreduce_shuffle.class/name valueorg.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和Driver3.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 MapperLongWritable, 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 ReducerText, IntWritable, Text, IntWritable { private IntWritable result new IntWritable(); Override protected 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); } }在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 groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.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 不想写JavaPython也能跑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端ShuffleCopy、Merge、ReduceReduceTask启动后会启动多个拉取线程同时从不同MapTask取回属于自己的分区数据。数据量不大时直接放在内存数据量大了就落盘。之后Reduce端会把从不同MapTask取回的数据进行归并排序最终形成一个“所有相同key连续排列”的有序迭代器这个迭代器就是传给reduce方法的values。这里的关键点是Reduce方法实际能看到的数据不只是它自己的MapTask而是所有MapTask中同一个分区的数据。所以无论你有多少个MapTask最终结果都不会丢数据。5.4 关键参数速查参数名作用常见设置mapreduce.job.reducesReduceTask数量默认1数据量大时调到5-10mapreduce.task.io.sort.mbMap端缓冲大小默认100MB大任务调大mapreduce.map.memory.mbMapTask容器内存默认1024MB堆内存不足就调大mapreduce.reduce.memory.mbReduceTask容器内存默认1024MBmapreduce.input.fileinputformat.split.maxsize输入分片最大值控制MapTask并行度mapreduce.job.map.output.collector.classMap输出收集器一般不用动调参的原则永远是先看默认值够不够不要一上来就乱调。大多数面试和实际场景只需要关注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查看syslogFileAlreadyExistsException输出目录已存在删除旧目录或换路径ClassNotFoundExceptionjar包没打全jar -tf检查重新打包连接被拒绝NameNode或ResourceManager没启动jps排查进程看启动日志StackOverflow或OOMMAP内存不足调大mapreduce.map.memory.mb结果全为空输入路径写错或文件为空hdfs dfs -ls检查输入目录进度卡在reduce 99%Reducer数太少或数据倾斜设多个Reducer考虑加Combiner最后再分享一点我自己跑WordCount的经验我在很多次测试里发现最快定位问题的方式不是反复盯代码而是把提交命令、输出目录、输入数据和日志这四样东西按顺序检查一遍。很多人一遇到报错就怀疑代码逻辑但实际上一半以上的问题集中在环境、路径和依赖上。如果你的目标只是把MapReduce跑通不需要追求一次性写出符合所有工程规范的代码只需按顺序走完编译、打包、提交、看日志、查结果这套流程。等这个闭环建立起来之后再逐步引入自定义计数器、Combiner、Partitioner、多输入格式这些进阶特性。最后再送一个实用小技巧调试阶段把输入文件刻意做得小一点比如只有几行内容。这样MapTask和ReduceTask各只有一两个日志短、跑得快任何错误都能迅速定位。等逻辑已经确认稳定了再换大数据量做性能验证。这是我用得最顺手的一个习惯。 
长河君 · 2026/10/22 23:23:54 · 阅读 1.2k Hadoop MapReduce实战：手把手带你跑通WordCount程序
刚接触Hadoop的人看到“wordMapper”“wordReduce”这种叫法其实对应的就是MapReduce里最经典的WordCount程序统计一堆文本中每个单词出现了多少次。它看起来简单但如果你能亲手把这段代码写完、编译、打包、提交到集群上跑通MapReduce的核心机制基本就打通了一半。这篇文章我按照平时真实跑任务的顺序来写从Mapper和Reducer怎么设计与实现到用什么命令提交、怎么读日志、踩了哪些坑一次讲透。适合第一次接触Hadoop、想把MapReduce程序真正跑起来的读者。1. 先想清楚WordCount的“MapReduce”到底在拆解什么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时代用JDK8Hadoop 3.x也支持JDK8但部分新功能需要JDK11。我的建议是选择一个稳定组合Hadoop 3.3.x 配 JDK8 或 JDK11这是目前最常用、遇到问题最容易搜到解决方案的组合不要一味追新。推荐使用Hadoop 3.3.6这类成熟版本。太老的2.7系列虽然教程多但和现代操作系统的兼容性越来越差太新的版本可能和某些生态组件还没适配。稳定压倒一切。2.3 解压、配置环境变量、准备启动脚本第一步下载Hadoop压缩包解压到某个目录假设是/opt/hadooptar -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 namefs.defaultFS/name valuehdfs://localhost:9000/value /property /configurationhdfs-site.xml里设置副本数伪分布式下副本数为1就够了configuration property namedfs.replication/name value1/value /property /configurationyarn-site.xml里启用YARN资源调度并设置NodeManager的辅助服务configuration property nameyarn.nodemanager.aux-services/name valuemapreduce_shuffle/value /property property nameyarn.nodemanager.aux-services.mapreduce_shuffle.class/name valueorg.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和Driver3.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 MapperLongWritable, 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 ReducerText, IntWritable, Text, IntWritable { private IntWritable result new IntWritable(); Override protected 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); } }在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 groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.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 不想写JavaPython也能跑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端ShuffleCopy、Merge、ReduceReduceTask启动后会启动多个拉取线程同时从不同MapTask取回属于自己的分区数据。数据量不大时直接放在内存数据量大了就落盘。之后Reduce端会把从不同MapTask取回的数据进行归并排序最终形成一个“所有相同key连续排列”的有序迭代器这个迭代器就是传给reduce方法的values。这里的关键点是Reduce方法实际能看到的数据不只是它自己的MapTask而是所有MapTask中同一个分区的数据。所以无论你有多少个MapTask最终结果都不会丢数据。5.4 关键参数速查参数名作用常见设置mapreduce.job.reducesReduceTask数量默认1数据量大时调到5-10mapreduce.task.io.sort.mbMap端缓冲大小默认100MB大任务调大mapreduce.map.memory.mbMapTask容器内存默认1024MB堆内存不足就调大mapreduce.reduce.memory.mbReduceTask容器内存默认1024MBmapreduce.input.fileinputformat.split.maxsize输入分片最大值控制MapTask并行度mapreduce.job.map.output.collector.classMap输出收集器一般不用动调参的原则永远是先看默认值够不够不要一上来就乱调。大多数面试和实际场景只需要关注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查看syslogFileAlreadyExistsException输出目录已存在删除旧目录或换路径ClassNotFoundExceptionjar包没打全jar -tf检查重新打包连接被拒绝NameNode或ResourceManager没启动jps排查进程看启动日志StackOverflow或OOMMAP内存不足调大mapreduce.map.memory.mb结果全为空输入路径写错或文件为空hdfs dfs -ls检查输入目录进度卡在reduce 99%Reducer数太少或数据倾斜设多个Reducer考虑加Combiner最后再分享一点我自己跑WordCount的经验我在很多次测试里发现最快定位问题的方式不是反复盯代码而是把提交命令、输出目录、输入数据和日志这四样东西按顺序检查一遍。很多人一遇到报错就怀疑代码逻辑但实际上一半以上的问题集中在环境、路径和依赖上。如果你的目标只是把MapReduce跑通不需要追求一次性写出符合所有工程规范的代码只需按顺序走完编译、打包、提交、看日志、查结果这套流程。等这个闭环建立起来之后再逐步引入自定义计数器、Combiner、Partitioner、多输入格式这些进阶特性。最后再送一个实用小技巧调试阶段把输入文件刻意做得小一点比如只有几行内容。这样MapTask和ReduceTask各只有一两个日志短、跑得快任何错误都能迅速定位。等逻辑已经确认稳定了再换大数据量做性能验证。这是我用得最顺手的一个习惯。