MapReduce基础编程实战：从WordCount到Shuffle与性能调优

我做了这么多年分布式计算相关的东西发现一个现象很多新人学MapReduce看两天教程就以为自己会了但是一上手写代码就懵。背了一堆API却不知道MapReduce到底解决的是什么问题每个阶段在干什么更别说遇到OOM、数据倾斜、小文件爆炸这些生产环境里的老大难问题。这篇文章我准备把MapReduce基础编程从头到尾捋一遍不讲虚的全是我实际写作业、调任务、填坑总结出来的实战经验。如果你正准备学Hadoop、要应付学校的分布式课程、或者刚入职需要维护公司的离线计算任务这篇文章都适用。我会从一个最简单的WordCount开始把MapReduce的编程模型、执行流程、Shuffle机制、常见坑、性能调优一次说透保证你看完能立刻动手写自己的第一个MapReduce作业并且知道遇到问题时该往哪个方向排查。1. MapReduce编程模型想明白这三件事才算入门学MapReduce之前我建议你先忘掉Hadoop忘掉那些繁琐的API。MapReduce本质上就解决一个问题如何把一个大任务拆成很多小任务并行处理然后再把结果汇总起来。它借鉴了函数式编程里面的Map和Reduce两个概念只不过把它用在了分布式数据处理的场景下。1.1 “分而治之”到底是怎么个分法举个生活中的例子。假设你要统计全公司一万个人的平均工资你会怎么做最笨的办法是把一万条数据都拉到你电脑上一个一个看。靠谱一点的做法是让每个部门的人力专员先把本部门的平均工资算好最后你再拿所有部门的平均值做加权平均。MapReduce就是后者。你的数据存放在集群的多台机器上Map阶段就是让每台机器“各管各的”只处理它本地的那份数据算出一个个中间结果。这些中间结果会按某种规则重新组织送到Reduce阶段也就是最后汇总的环节。在分布式系统里MapReduce对程序员屏蔽了并行计算、数据分发、容错、负载均衡这些复杂问题。你只需要写两个函数map函数负责把一条数据转换成(key, value)对reduce函数负责把相同key的value批量汇总。至于集群怎么调度、数据怎么传输、某台机器挂掉了怎么办通通不用你操心框架全包了。1.2 数据中心不移动移动的是计算这是MapReduce设计哲学里极其重要的一点。数据存放在HDFS上每个Block分布在不同的DataNode节点上。如果你的代码要在集群上跑框架的做法是把jar包和计算逻辑发送到数据所在的节点而不是把几TB的数据通过网络搬到一台机器上。这个设计在早期分布式计算里非常关键。网络传输的成本远高于本地磁盘IO和CPU计算尤其是数据量达到TB级别之后“计算跟着数据走”不是优化建议而是必须遵守的规则。所以你在提交MapReduce作业时通常要把jar包放到HDFS上或者在提交客户端上指定框架会优先在数据本地启动任务。1.3 一个作业就是一个完整的DAG严格来说MapReduce作业的执行过程是输入分片(InputSplit) - Map任务 - Shuffle(分区、排序、合并) - Reduce任务 - 输出。整个过程可以被看作一个简化的有向无环图每个Map任务处理一个或多个分片多个Map任务并行执行它们各自输出中间结果后框架按分区规则把数据路由到对应的Reduce任务。这也就解释了为什么MapReduce不适合做迭代式计算。图计算、机器学习这类需要反复迭代、跨阶段传递数据的任务每个迭代都意味着一次完整的MapReduce流程落盘、读取、序列化、反序列化开销巨大。所以后来才有了Spark、Flink这类基于内存计算的框架。2. 手写WordCountMapReduce的Hello World完整拆解要说MapReduce基础编程WordCount就是绕不开的入门题。它虽然简单但五脏俱全——输入格式化、Map逻辑、Reduce逻辑、输出格式化、作业配置全部涉及。我把完整代码写出来一步一步拆解给你看。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.LongWritable; 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 { // 继承Mapper类四个泛型参数分别是 // MAPINKEY: 输入键类型这里是行偏移量用LongWritable // MAPINVALUE: 输入值类型这里是每一行的内容用Text // MAPOUTKEY: 输出键类型这里是一个单词用Text // MAPOUTVALUE: 输出值类型这里是单词出现次数用IntWritable public static class TokenizerMapper extends MapperLongWritable, Text, Text, IntWritable { private final static IntWritable one new IntWritable(1); private Text word new Text(); // map方法是核心逻辑每读取一行文本就调用一次map Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // StringTokenizer是Java自带的字符串分割器按空白字符分割 StringTokenizer itr new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); // 每切出来一个单词就输出一个 (单词, 1) context.write(word, one); } } } // 继承Reducer类四个泛型参数分别是 // REDUCEINKEY: 输入键类型对应Map的输出键类型 Text // REDUCEINVALUE: 输入值类型对应Map的输出值类型 IntWritable // REDUCEOUTKEY: 输出键类型 Text // REDUCEOUTVALUE: 输出值类型 IntWritable public static class IntSumReducer extends ReducerText, IntWritable, Text, IntWritable { private IntWritable result new IntWritable(); // reduce方法在框架中是这样调用的 // 框架把Map输出的所有数据按key分组同一个key的所有value构成一个迭代器 // 每切换一个key就会调用一次reduce Override protected void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException { int sum 0; // 把所有value加起来就是这个单词在全部数据中出现的次数 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); // Combiner是什么、为什么在这里加上后面我会专门讲 job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); // 输出key的类型 job.setOutputKeyClass(Text.class); // 输出value的类型 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); } }2.1 为什么Hadoop不用Java基本类型你有没有注意到上面代码里用的都是LongWritable、Text、IntWritable这种类型而不是long、String、int。这不是故意绕弯子而是Hadoop设计的一层序列化体系。MapReduce执行过程中数据要跨节点传输、要写磁盘、要经过网络这时候数据必须被序列化成字节流。Java自带的Serializable接口序列化出来数据太重了包含大量的类信息、继承结构信息在网络传输和磁盘存储中开销很大。Hadoop自己实现了一套序列化机制叫Writable它只保留数据本身的内容非常紧凑。Text和String的区别也值得注意。Text是UTF-8编码的可变长度字节序列它不等于Java的String用Text做key时比较、排序、哈希都是基于字节内容的效率比String高不少。我见过有人为了省事直接在Mapper里toString()转成String操作然后又new Text()转回来数据量小没问题数据量大了性能和内存都会很难看。2.2 一个MapReduce作业的完整生命周期光知道写代码还不够你得理解你写的这些类、配置在提交到集群之后到底经历了什么。我用大白话把流程串一遍第一步客户端先把作业的jar包、配置文件和依赖上传到HDFS上一个以作业ID命名的临时目录。第二步客户端向ResourceManager提交作业ResourceManager会创建一个Job然后向客户端返回一个用于监控进度的链接。第三步ResourceManager在某个NodeManager节点上启动ApplicationMasterAMAM负责这个作业的全部调度和管理。AM向ResourceManager申请一组容器根据输入的数据分片数决定启动多少个Map任务。第四步每个Map任务读取一个分片的输入数据经过InputFormat的RecordReader解析成(key, value)对逐条调用你的map()方法。这里的key默认是行偏移量value是整行文本。你的map方法输出新的(key, value)框架会先写入内存缓冲区。第五步Map任务跑完后作业进入Shuffle阶段。框架把Map输出按key分区每个分区对应一个Reduce任务。分区内的key会排序如果有Combiner会先做一次本地合并然后数据落到Map节点的本地磁盘。第六步所有Map任务完成AM开始启动Reduce任务。Reduce任务从每个Map节点拉取属于自己分区的数据合并、排序然后逐键调用你的reduce()方法。第七步Reduce输出最终写入HDFS整个作业结束。HDFS上的临时目录会被清理任务日志保留下来。这个过程看起来简单但对排查问题特别重要。比如你发现作业卡在“Reduce状态一直显示Running”大概率是某几个Reduce任务在拉取数据时卡住了或者Map任务有数据倾斜某个分区的数据量远超其他分区。2.3 提交作业时那些容易翻车的细节代码写对了不代表能跑通提交作业才是第一个大坑。我帮你把最容易翻车的几个细节列一遍输出目录绝对不能预先创建。FileOutputFormat要求输出目录不存在如果存在会直接抛FileAlreadyExistsException。输入文件如果是一堆小文件性能惨不忍睹。默认的TextInputFormat每个小文件都可能单独成一个分片每个分片启动一个Map任务启动开销远大于处理开销。本地跑和集群跑的环境不一样。本地用IDEA跑时Configuration会加载你本机的core-site.xml这类配置文件可能找不到hadoop相关环境在集群上跑的时候最好用job.getConfiguration()去设置参数并确保setJarByClass填的是你主类所在的全限定名否则集群找不到你的Mapper、Reducer类。内存相关参数要提前设好。如果是大作业建议在作业提交脚本里加上mapreduce.map.memory.mb、mapreduce.reduce.memory.mb、mapreduce.map.java.opts等配置否则默认1GB内存很容易让巨型Map任务OOM。3. Shuffle机制不弄懂写十个作业还是门外汉我说句实在话MapReduce的代码谁都能写Map函数和Reduce函数本身都很简单真正决定你作业能不能跑快、能不能跑稳的是中间的Shuffle。很多人在面试或者做性能优化的时候一问Shuffle就含糊其辞原因很简单——Shuffle是Mapper和Reducer之间所有数据处理步骤的总称它不体现在你写的代码里而是框架自动做的大家看不见摸不着自然难以理解。3.1 Map端的Shuffle缓冲区、环形内存与溢写Map任务的输出不是直接写磁盘的而是先写进一个内存缓冲区。这个缓冲区默认大小是100MB可以通过mapreduce.task.io.sort.mb配置。缓冲区分成等大小的两部分每部分约50MB分别用于存放数据和索引。当缓冲区写满80%mapreduce.map.sort.spill.percent默认0.8时后台线程就开始执行溢写spill操作。溢写的过程是这样的先把缓冲区中的数据按分区进行排序再按分区合并然后考虑是否执行Combiner如果需要Partitioner做二次分区也会在这里做最后把排序后的数据写入一个小文件。注意溢写时Map任务并没有停止它会继续往缓冲区里写数据。如果溢写速度跟不上数据生成速度缓冲区满了Map任务就会阻塞等待直到溢写完成。整个Map任务跑完可能已经产生了多个溢写文件。框架会在最终阶段把这些小文件合并成一个大文件并且合并时依然会排序。所以Map端的输出本质上是“每个分区内有序”的一系列文件。我做性能调优的时候最常做的事情就是调大mapreduce.task.io.sort.mb。如果Map任务处理的是大量文本行、输出的键值对很密集默认100MB的缓冲非常容易频繁溢写每次溢写都是一次磁盘IO和排序。调到300MB或400MB能明显减少溢写次数。前提是Map任务的堆内存要相应配大比如mapreduce.map.memory.mb设置到2GBmapreduce.map.java.opts设置-Xmx为1.6GB左右留下JVM本身和其他开销的空间。3.2 Reduce端的Shuffle拉取、归并、再排序Reduce端Shuffle分为三个阶段拉取Fetch、合并Merge、归约排序。Reduce任务在启动后会先向AppMaster获取已完成的Map任务的状态和它们输出分区的物理位置然后启动若干个拉取线程从各个Map节点上拷贝属于自己分区的数据。拉取的数据先进入Reduce端的内存缓冲区缓冲区不够了会写到磁盘。多份数据拉到本地后Reduce会对它们进行“归并”——不是简单的拼接而是将多份有序的文件合并成一个有序的大文件合并时如果开启Combiner也会再执行一次。这里有个特别容易忽略的知识点Reduce端对数据的处理是按key分组的。归并排序完成后框架把数据按key分成一组一组每一组数据的key是相同的对应的value组成一个迭代器调用一次reduce()方法。3.3 Partitioner决定每条数据去找哪个ReduceShuffle里还有一个隐藏角色叫Partitioner它的职责是决定Map输出的每一条(key, value)应该被分配到哪个Reduce任务。默认实现是HashPartitioner它做的事情只有一行代码return (key.hashCode() Integer.MAX_VALUE) % numReduceTasks;先用key.hashCode()取哈希然后把符号位去掉保证非负再对Reduce任务数取模。这意味着如果Reduce任务数是固定的那么相同key的哈希值相同最终一定进入同一个Reduce分区这样reduce()才能拿到这个key的全部value做汇总。如果业务上有特殊需求比如想让某个特定key的数据进入指定的Reduce或者想打破默认的哈希分布逻辑你需要自己实现Partitioner类。我写过一次自定义Partitioner场景是把所有“异常”状态的数据集中到一个Reduce里人工核查剩余正常数据平均分给其他Reduce。掌握这个钩子之后你对Shuffle的掌控能力才算是真正上了一个台阶。3.4 CombinerMap端的“小Reducer”Combiner是另一个容易被忽视的组件。它本质上是一个运行在Map端的“预Reducer”对Map输出的局部数据进行一次合并从而减少Map输出落盘的数据量和跨节点传输的数据量。但它有非常严格的限制Combiner的输入输出类型必须与Mapper的输出类型一致而且combine操作必须是可重复执行的、幂等的操作不能破坏最终的Reduce结果。以WordCount为例(hello, 1)、(hello, 1)、(hello, 1)在Map端先用Combiner合并成(hello, 3)Reduce端再和来自其他Map的(hello, 2)合并成(hello, 5)最终结果正确。但如果你写的是求平均值的逻辑Map端合并成平均值Reduce端再平均结果就会严重失真。这种场景就不能用Combiner或者需要换一种合并策略。判断一个操作能不能做Combiner最核心的标准是它在局部合并后再全局合并结果和全量合并完全一致。求和、求最大值、求最小值、计数、拼接字符串都满足求平均值不满足。4. 从能跑到跑得快性能调优的实战思路把作业跑通只是及格线在真实生产环境里数据量大、业务要求时效高随便一个“能跑”的作业可能慢到让人崩溃。我总结了自己做MapReduce调优的几个方向每个方向都踩过坑分享出来给你参考。4.1 数据倾斜处理99亿条日志也不怕数据倾斜是MapReduce作业最常见的杀手。表现形式是作业进度条卡在99%一堆Reduce任务只剩一两个还在运行而且那一个Reduce任务要处理的数据量是其他Reduce任务的几十倍甚至上百倍。原因很直接某个key的分布太集中所有数据都哈希到了同一个分区。比如统计日志里某个特定IP的访问量这个IP的数据量可能占了全部日志的30%那么负责这个IP分区的Reduce就必然慢。处理数据倾斜的思路大概有四种增加Reduce并行度mapreduce.job.reduces调大之后每个key的哈希取模分到不同Reduce的概率增大但解决不了“绝对热点”。Map端Combiner预处理如果能在Map端先把热点key的部分数据进行聚合传输给Reduce的数据量会大幅度减少。两阶段聚合局部加盐再聚合这是我最常用的方案。Map阶段对key加上随机数前缀让原本集中的key“打散”到多个Reduce里做局部聚合然后再启动第二个作业去掉前缀、做最终聚合。代价是跑了两个作业但在极端倾斜场景下效果立竿见影。自定义Partitioner手动把热点key单独拿出一个Reduce其余正常数据平均分到其他Reduce。4.2 小文件问题与输入格式的选择HDFS的设计定位是大文件存储默认Block大小是128MB。当你的输入目录里有成千上万个几十KB甚至几KB的文件时每个文件很可能单独成为一个InputSplit每个Split对应一个Map任务。于是几十万个Map任务同时启动ResourceManager光分发任务、回收任务就能拖垮整个集群。解决方案有两个层面。数据源头尽量把多个小文件合并成大文件比如定期用SequenceFile或者CombineTextInputFormat来读取输入。CombineTextInputFormat可以从业界的角度说是一个非常实用的类它能把多个小文件合并成一个Split从根本上减少Map任务数。job.setInputFormatClass(CombineTextInputFormat.class); CombineTextInputFormat.setMaxInputSplitSize(job, 67108864); // 64MB设置了一个合理的maxInputSplitSize之后原本几千个Map任务能降到几十个集群负载立刻正常。这个参数在HDFS上文件数量庞大时是必调的。4.3 压缩和序列化的选择在中间数据和最终输出阶段都建议开启压缩。MapReduce中间数据默认是不压缩的数据量大时Shuffle传输量惊人。我习惯在mapred-site.xml或作业配置里开启mapreduce.map.output.compress压缩算法选Snappy或LZOCPU开销低、压缩比也不错。property namemapreduce.map.output.compress/name valuetrue/value /property property namemapreduce.map.output.compress.codec/name valueorg.apache.hadoop.io.compress.SnappyCodec/value /property最终输出要不要压缩取决于下游系统。如果下游是Hive分析或者后续还是要做MapReduce压缩能减少读取时间和存储空间如果下游是实时系统或需要直接读文本那可能就要权衡一下。4.4 调整并行度与资源配额Map任务的并行度理论上等于输入分片的数量Reduce任务的并行度由mapreduce.job.reduces指定默认是1。很多人不知道这个默认值写了一个Reduce跑完整个作业数据量几亿条慢得感人。但Reduce数量也不是越大越好。每个Reduce任务都有启动、拉取、合并、清理的固定开销Reduce太多反而会让每份数据太小浪费调度时间。一个可参考的经验公式是max(1, min(集群CPU核数 * 2, 输入数据量 / 每Reduce期望处理量(约1GB)))。我一般根据数据规模把Reduce控制在50到200之间然后看CPU利用率和运行时长再微调。5. 排查MapReduce作业问题的思路与方法跑MapReduce作业一天不出点问题是运气好出问题是常态。我把这些年踩过的问题分几类讲顺便把排查链路完整说出来。按照这个思路走大部分问题你都能独立定位。5.1 作业卡住不动的排查链路先看日志。在YARN的ResourceManager页面上找到你的Application ID进入ApplicationMaster的日志界面看每个任务的运行状态。如果出现LocalizedResource卡住或者任务一直处于PENDING状态第一反应是资源不够。此时看集群队列的资源使用率确认是不是你所在的队列资源被别人占满了。如果任务是RUNNING但进度永远是0%多半是Map任务读取输入有问题。输入文件是压缩格式却没有对应的解压器或者文件路径写错导致没有Split生成。我遇到过一种情况输入目录下有大量_SUCCESS文件这些文件是Hive生成的会被FileInputFormat过滤掉倒不影响但如果你自定义了输入格式过滤逻辑没有处理好就会导致任务永远在等待输入。5.2 运行结果不对的排查思路结果不对比跑不动更让人头疼因为代码看起来都对、作业也跑完了就是数据对不上。我通常从三个方向检查Combiner破坏语义检查Combiner的操作是否满足幂等和可交换性。Partitioner自定义逻辑有缺陷检查getPartition里是否有某种key没有覆盖到、返回值有没有超出Reduce个数范围。Reduce里对Iterable的误用这是新手特别容易踩的坑。Java的Iterable只能遍历一次你第一遍循环计算了总和第二遍循环再想取其他值会发现什么都没有了。所以如果需要遍历多次得先拷贝到本地List。5.3 OOM与资源不足的常见解法Map任务和Reduce任务都有自己的内存上限。如果Map任务OOM第一步先看mapreduce.map.memory.mb设置了多少再结合mapreduce.map.java.opts里的-Xmx确认堆大小小于容器总内存。其次是看Map处理的单条数据有多大如果数据里有几百MB的大字段建议单独处理。Reduce阶段OOM常见原因是Reduce端拉取的数据在内存里堆积。解决方案是把mapreduce.reduce.shuffle.input.buffer.percent调低让更多数据直接写磁盘或者调大mapreduce.reduce.memory.mb。还有一个冷门原因就是final reduce阶段对全量数据做了不合理的缓存比如把所有数据都塞进了一个HashMap再逐条处理这种写法无论怎么调参都会炸改逻辑才是根本。5.4 日志是排查的根MapReduce的日志分三类stdout、stderr和syslog。系统诊断信息都在syslog里stderr通常也是异常堆栈。如果你在Mapper或Reducer里打了println输出会进stdout但因为日志量大的时候会被截断或轮转我建议用context.getCounter加自定义计数器来观察数据量变化这种方式可观测性更强对排查数据倾斜、统计异常值特别有效。6. MapReduce技术的演进以及对基础编程学习的影响写到这里可能有人会问现在Spark、Flink这么火Hadoop MapReduce是不是已经过时了我的观点很明确MapReduce没有过时它是理解大数据计算的一把钥匙。Spark的RDD、DAG、Shuffle很多概念是对MapReduce模型的扩展和优化。你把MapReduce的Shuffle彻底搞明白了再去学Spark的宽窄依赖、Shuffle机制会发现知识是连贯的甚至有些Spark的调优思路可以直接平移过来。当然MapReduce在实时性上的确差一些因为它每一步都要落盘延迟高。但它有一个得天独厚的优势稳定性极高可以处理PB级数据运维形态也很成熟。在很多公司的离线数仓、凌晨批量计算场景里MapReduce依然在服役尤其是Hive on MapReduce这种组合跑得又稳又省心。对于学生和刚转行的人来说我建议一定亲手在本地或者测试集群上跑通MapReduce的入门作业。你不需要用多么复杂的数据集把官方示例WordCount跑通然后再写一个问题统计用户访问日志里每天Top10的URL。这个题目涉及时间维度分组、全局排序、TopN裁剪做完之后你对MapReduce基础编程、二次排序、自定义Writable的感觉会完全不同。我给朋友带新人的时候总说这句话MapReduce编程难的从来不在地图函数和Reduce函数怎么写而在你能不能想清楚数据在每一个阶段经历了什么。当你看着任务进度条能想象出数据正在缓冲区里排序、正在被分区、正在从某个节点拉取你就真的入门了。说起来我当年刚开始接触分布式计算时也被Reduce阶段那个“一个key调一次reduce”的设计绕晕过。直到有一次在生产环境排查数据倾斜盯着Web UI上那一个孤零零的、跑了三小时还没结束的Reduce任务才彻底理解了Partitioner、分组和Shuffle之间的关系。从那以后我每写一个MapReduce作业都会先问自己三个问题每个key对应多少数据量有没有可能倾斜Combiner在这里能不能用这三个问题想清楚了作业基本就坑少了一大半。希望这篇超长的实战总结也能帮你少走几条弯路。