第一次跑通 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是该行在整个文件中的偏移量字节数类型是LongWritablevalue是这一行的文本内容类型是Text这也是为什么 WordMapper 的泛型签名必须是MapperLongWritable, Text, Text, IntWritable的原因。有同学问过为什么第一个参数不能用Integer、第二个不能用String原因在于 Hadoop 的分布式序列化机制。Java 自带的序列化会把整个对象图写进字节流冗余信息太多所以 Hadoop 定义了自己的序列化类型LongWritable、Text、IntWritable这些 Writable 类型体积小、可高效序列化在跨网络传输时开销远低于原生 Java 类型。这是初学者最容易忽略但面试也最常被问到的点。输出格式默认是TextOutputFormatReducer 每context.write(key, value)一次输出文件里就多一行key Tab value。最终文件会以part-r-xxxxx命名r表示 reducer 的输出xxxxx是 reducer 的编号。看到这个文件说明整个作业成功了。2. 逐行拆解WordMapper、WordReducer 和驱动类2.1 WordMappermap 方法输入的 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 MapperLongWritable, 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 WordReducerreduce 方法拿到的 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 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); } }这里初学者最容易犯的错是想把values转成List再遍历。不要这么干。values是一个延迟迭代的Iterable它背后可能是一大堆来自不同 Map 任务、通过网络传输过来的数据。如果数据量大你把它全部缓存在内存里轻则内存溢出重则拖垮整个 NodeManager。正确做法就是在这里用一次遍历直接聚合完。Reducer 的输入是已经排好序的同一个 key 的所有 value 会被分成一组调用一次reduce()组的顺序按照 key 排序。也就是说如果你有 10 万个单词reduce()方法会被调用 10 万次每次处理一个单词的所有计数。另外注意输出的 key 类型是Textvalue 类型是IntWritable这和 Mapper 的输出类型是一致的。Shuffle 阶段搬运的就是 Mapper 的key, value对所以Mapper 的输出 key/value 类型必须和 Reducer 的输入 key/value 类型完全匹配否则运行到一半会报类型转换错误。2.3 驱动类配置 Job 的三大核心动作有了 Mapper 和 Reducer还需要一个驱动类把它们串起来提交给集群。这也是WordReduce这类描述里最容易漏掉的部分——很多人以为写完 Mapper 和 Reducer 就完了实际上没有 driverHadoop 根本不知道你写的是什么。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, LongWritableReducer 最终输出的是Text, IntWritable就必须显式补充setMapOutputKeyClass(Text.class)和setMapOutputValueClass(LongWritable.class)否则运行到 map 输出阶段就会因为类型不匹配而报错。还有一个小细节job.setCombinerClass(WordReducer.class)这一行的注释我特意保留了。Combiner 是可选项但它对性能影响极大后面第五章会专门讲。这里先记住一个原则Combiner 的逻辑必须满足交换律和结合律否则结果会出错。词频累加天然满足所以可以直接复用 Reducer 类。3. 从代码到集群运行编译、打包、提交的完整链路3.1 Maven 依赖与打包注意事项MapReduce 程序要用 Maven 管理依赖最核心的依赖就一个dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.3.6/version scopeprovided/scope /dependencyscope设为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 输入数据准备先上传到 HDFSMapReduce 读取的不是本地文件系统的文件而是HDFS 上的文件。这一步是很多初学者在本地试运行时报文件找不到的根源。准备一个测试文件内容随便写几行英文例如hello world hadoop hello mapreduce hadoop mapreduce wordcount demo然后上传 HDFShdfs 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 mapsMap 任务数量。一个输入分片对应一个 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 ClassNotFoundExceptionJar 没打全或不认识入口类这个报错也有几种不同表现报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 MapDriverLongWritable, 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可以实现一个自定义PartitionerText, IntWritableOverride 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.mb2048 \ -Dmapreduce.reduce.memory.mb3072 \ input output第二压缩输出数据。HDFS 上存文本文件非常浪费空间MapReduce 输出可以启用压缩-Dmapreduce.output.fileoutputformat.compresstrue \ -Dmapreduce.output.fileoutputformat.compress.codecorg.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 生效与否对指标的影响。把这些体感建立起来之后再看任何复杂的分布式计算框架你都会比那些只背概念的人多一层它底层到底怎么跑的直觉。
