接触大数据的人绕不开的第一个编程框架就是MapReduce。不管是在实训平台刷基础编程、做HDFS与MapReduce综合实训还是处理招聘数据清洗这种综合案例你都会发现题目看上去五花八门真正用到的核心知识其实高度集中。这篇内容我打算把MapReduce从“任务如何被拆开执行”讲起串起编程模型、排序分组、倒排序索引和清洗案例最后给出一张异常排查参考表适合正在刷实训关卡、准备面试或者刚接手集群作业的同学。1. 先搞清楚MapReduce到底在解决什么问题1.1 核心思想把大任务拆成可以并行的两段MapReduce最核心的思想是“分而治之”。大数据场景下的计算任务本质上可以分成两类一类是单条记录可以独立完成的比如给一万行日志做格式清洗另一类必须跨记录聚合比如统计全国所有招聘岗位的平均薪资。Map阶段解决前者Reduce阶段解决后者。我用一个生活类比解释假设要统计一栋楼里每个单元有多少人。Map阶段相当于每层楼各派一个管理员各自把自己负责楼层的人数数出来登记在一张小纸条上Reduce阶段相当于一楼大厅的总台把各层送上来的纸条汇总相加。这样楼层数越多只需要加人每个楼层的工作互不干扰整体效率接近线性提升。这个设计解决了传统单机程序处理TB级数据时“内存放不下、计算时间太长”的问题。MapReduce把输入数据切碎后分配到多台机器上并行处理所有中间结果都以键值对形式在网络间传递框架统一处理传输、排序、容错和负载均衡。你只需要把精力放在业务逻辑上写清楚Mapper怎么处理一条记录、Reducer怎么聚合一组记录剩下的事框架包了。1.2 从提交到出结果一个任务的完整生命周期理解MapReduce的运行流程比背概念重要得多。一个Job从提交到结束大概要经过这么几个环节客户端把jar包、输入路径、输出路径和配置打包提交给资源调度器。框架根据输入目录调用InputFormat把文件切分成若干split。每个split对应一个Map任务TextInputFormat默认是“按字节偏移量加一行内容”的方式读取数据。Map任务逐行处理输入把计算结果写到内存环形缓冲区。缓冲区有大小上限默认100MB达到阈值后触发溢写spill溢写前会先分区、再排序。每个Map任务可能产生多个spill文件最终会归并成一个大的输出文件。Reduce端从这个输出文件里拉取属于自己分区的数据做归并排序把相同key的value排到一起。Reduce拿到分好组的键值对执行reduce方法结果直接写入HDFS上的输出目录中生成part-r-00000这类文件。这里有个容易忽略的细节Map任务的数量不是“你写了多少行Mapper”而是输入文件被切成了多少split。一个128MB的数据块通常对应一个split如果你放了一堆几十KB的小文件每个小文件也可能单独占一个Map任务导致大量时间和资源消耗在启动和调度上而不是真正计算。所以做综合实训前尽量把数据合并成大文件再丢到HDFS上能少踩很多坑。1.3 HDFS和MapReduce为什么总被放在一起讲实训里经常出现“HDFS和MapReduce综合实训”这种关卡很多人以为这是两个独立知识点其实它们是存储和计算的搭档关系。HDFS负责存储把大文件拆成128MB的块副本默认3份分散在不同机器上解决“数据往哪儿放”的问题。MapReduce负责计算基于HDFS上的数据做计算解决“数据怎么算”的问题。两者协作最关键的优化叫“数据本地性”Map任务启动时框架优先把它调度到数据所在节点上直接读本地磁盘而不是跨网络拉数据。用一句话概括就是“数据不动计算动”计算代码很小传输一次可以接受数据很大来回拷贝代价太高。这也是HDFS适合跑批任务、不适合低延迟查询的原因。做综合实训时标准的操作顺序是先用hdfs dfs -mkdir创建目录用hdfs dfs -put上传数据文件确认数据落盘再提交MapReduce任务。很多人喜欢本地调试时直接用本地文件路径上传之后却忘了改输入路径导致提交到集群上后报“输入路径不存在”这种问题几乎每个班都会遇到几次。2. 核心编程模型Mapper、Reducer和Driver2.1 Mapper一行输入对应一个输出点Mapper是所有任务的起点。它的输入是键值对通常key是行偏移量LongWritablevalue是整行文本Text。你在map方法里写业务逻辑处理完调用context.write输出新的键值对。以最经典的WordCount为例public static class WordCountMapper extends MapperLongWritable, Text, Text, IntWritable { private Text outKey new Text(); private IntWritable outValue new IntWritable(1); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString(); if (line null || line.trim().isEmpty()) { return; } String[] words line.trim().split(\\s); for (String word : words) { outKey.set(word); context.write(outKey, outValue); } } }这里面有个很实用的习惯尽量不要在map方法内部用new创建输出对象比如context.write(new Text(word), new IntWritable(1))。因为一条大文件可能有一亿行每一行都new两个对象会给GC带巨大压力。改成类成员变量复用是大量老工程师的常规做法。Mapper输出的类型决定了下游shuffle阶段的序列化方式所以map方法里面写什么key、什么value要和类签名完全一致否则运行到一半就抛ClassCastException。这个坑在实训里出现频率非常高。2.2 Reducer按key聚合的归并逻辑Reducer的输入是框架处理好的结果键是分组后的key值是同一个key对应的所有value的迭代器。你只需要写清楚“拿到这个key和一批value输出什么”。在reduce方法里有一个非常经典的翻车现场有人想把value列表存下来做二次处理写了类似values.forEach(v - list.add(v))的代码结果发现最后list里全是同一个对象数据全变成了最后一条。原因是框架会复用同一个value对象每次迭代时把它替换成新值引用本身没有变。要保存数据必须复制一份比如list.add(new Text(v))或者像WordCount一样直接累加数值public static class WordCountReducer extends ReducerText, IntWritable, Text, IntWritable { private IntWritable outValue new IntWritable(); Override protected void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException { int sum 0; for (IntWritable value : values) { sum value.get(); } outValue.set(sum); context.write(key, outValue); } }Reducer的数量决定了输出文件的个数也直接影响到最终结果是否可能倾斜。默认每个分区由一个Reduce任务处理分区数设置多少需要结合业务量考虑不是越大越好。2.3 Driver三个关键配置决定任务成败Driver类在实训里经常被平台预先写好但面试或实际项目中你迟早要自己写。它做的事情主要是配置Job并提交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.setReducerClass(WordCountReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); }这里最重要的就是三组配置setMapOutputKeyClass和setMapOutputValueClass决定了Map端输出类型。setOutputKeyClass和setOutputValueClass决定了Reduce端输出类型。如果Map端和Reduce端类型恰好一致第二组可以省略但写上更稳妥避免踩坑。还有一个容易忽略的习惯本地调试时fs.defaultFS和mapreduce.framework.name的配置没有改过来就会莫名其妙去连集群然后报连接拒绝。这种情况在Windows本地上尤其常见因为Hadoop的本地模式在Windows上还经常报缺少winutils.exe的错处理成本并不低。建议实训阶段尽量在虚拟机、容器或集群环境上跑省去一堆环境兼容问题。2.4 容易踩坑的类型系统与序列化Hadoop没有直接使用Java的Integer、String作为MapReduce的键值类型而是用自己的IntWritable、Text等。原因在于Java原生序列化太“重”一个字符串序列化后还带着对象头、类信息等大量冗余在网络传输和磁盘写入时效率很低。Hadoop实现了自己的Writable协议序列化结果精简传输快还能保证跨Java进程解析一致。常用的类型映射表看一眼就有数Java类型Hadoop Writable类型适用场景StringText文本内容、词语、路径IntegerIntWritable计数、序号、平均薪资的整数部分LongLongWritable偏移量、时间戳、大数值DoubleDoubleWritable浮点数、薪资均值BooleanBooleanWritable标记位Map/List不支持直接传输需要拆成多个键值对或用Text拼接如果业务需要把一个对象整体当作key或value就必须自定义类实现Writable接口并实现write和readFields方法。这个知识点在下面的自定义排序里是刚需先记住一个结论凡是自定义Bean一定要提供无参构造器因为框架需要反射创建对象没有无参构造器运行时会直接报错。3. 高频考点排序与分组到底怎么实现3.1 默认排序行为与数值排序陷阱MapReduce的shuffle阶段自带排序默认按key的升序排列。但这个“升序”对于不同的Writable类型含义不同Text比较的是字典序IntWritable比较的是数值大小。很多人踩过的坑是把“100、20、3”这些数字写成了Text类型结果排序结果变成了“100、20、3”或某种字典序排列而不是数值升序。要按数值排序要么把key用IntWritable/LongWritable封装要么在自定义Bean的compareTo里把String转成数字后再比较。默认排序虽然存在但它只能保证“每个Reduce内有序”不能保证“全局有序”。要想让所有输出数据整体有序通常只指定一个Reduce但这样会丢掉并行度或者自己实现TotalOrderPartitioner那是个进阶话题。实训里的大多数关卡用不到全局排序理解到这里就够了。3.2 自定义排序用WritableComparable定义你的比较规则实训里常见的一类题目是“自定义排序”例如按照部门分组组内按薪资从高到低排序。默认的key是单一字段实现不了这种复合规则所以需要把部门和薪资封装成一个自定义Bean实现WritableComparable接口。public class EmployeeBean implements WritableComparableEmployeeBean { private String dept; private double salary; public EmployeeBean() { } public EmployeeBean(String dept, double salary) { this.dept dept; this.salary salary; } Override public int compareTo(EmployeeBean o) { int deptDiff this.dept.compareTo(o.dept); if (deptDiff ! 0) { return deptDiff; } return Double.compare(o.salary, this.salary); } Override public void write(DataOutput out) throws IOException { out.writeUTF(dept); out.writeDouble(salary); } Override public void readFields(DataInput in) throws IOException { this.dept in.readUTF(); this.salary in.readDouble(); } public String getDept() { return dept; } public double getSalary() { return salary; } }细节解释一下compareTo返回负数、零、正数分别表示当前对象小于、等于、大于比较对象。我这里返回Double.compare(o.salary, this.salary)是把比较方向倒过来实现薪资降序。写write和readFields时字段顺序必须一一对应先写String再写Double读取时也要先读String再读Double。顺序错乱数据反序列化后全乱套这种bug定位起来非常隐蔽。在Driver里把Map输出key改成这个BeanReduce就能按“部门分组、组内薪资降序”拿到有序数据。很多时候自定义排序题的投资回报率很高理解了这一个 Bean 的套路能解决一大类题目。3.3 分组排序改造reduce分组的GroupingComparator“分组排序”是实训里的重难点很多同学会把它和3.2的排序Bean搞混。先说清楚概念排序、分区、分组是shuffle阶段三件不同的事。排序决定的是同一个Reduce内部所有键值对按什么顺序排列。分区决定的是每个key进入哪个Reduce。分组决定的是Reduce方法被调用的次数也就是哪些key的value会被合并到同一个迭代器里。默认情况下分组规则就是“key完全相等”。但有些业务需要“多个不同的key当成一组处理”。典型场景订单明细表里每条记录包含订单号和金额要求找出每个订单金额最高的前两条记录。这时排序规则应该是“订单号升序 金额降序”但分组规则只看订单号即使订单号相同而金额不同也要进入同一个reduce方法。实现思路是自定义OrderBeancompareTo里先比订单号、再比金额降序自定义Partitioner保证同一个订单号所有数据进同一个Reduce再自定义GroupingComparatorreduce时的分组只看订单号。public class OrderGroupComparator extends WritableComparator { protected OrderGroupComparator() { super(OrderBean.class, true); } Override public int compare(WritableComparable a, WritableComparable b) { OrderBean oa (OrderBean) a; OrderBean ob (OrderBean) b; return oa.getOrderId().compareTo(ob.getOrderId()); } }然后在Driver里设置job.setGroupingComparatorClass(OrderGroupComparator.class);reduce方法只需要取第一个value就是金额最大的记录取前两个就是Top2。框架在调用reduce前已经把同组数据排序好所以取第一个值这个操作是安全的。顺带提醒一个构造器细节继承WritableComparator时super(OrderBean.class, true)的第二个参数表示框架会自动创建OrderBean实例来做比较这个参数很容易被漏掉漏掉以后会报反射创建对象的异常。3.4 倒排序索引一个接近真实业务的完整示例倒排序索引这个名字听着高大上本质就是一个“词语到文档列表”的映射。搜索引擎要回答“哪些网页包含某个关键词”靠的就是这张表。MapReduce实训里的“倒排序索引”关卡会让你用并行计算的方式构建这个映射。我给出一个可运行的思路。假设输入是多个文件每个文件若干行文本最终输出格式是word1 --- doc1.txt: 3, doc2.txt: 1 word2 --- doc1.txt: 2Map端要同时知道“单词”和“它来自哪个文件”所以输出key用“单词文件名”拼接value给1方便后面统计词频。public static class InvertIndexMapper extends MapperLongWritable, Text, Text, Text { private Text outKey new Text(); private Text outValue new Text(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString(); if (line null || line.trim().isEmpty()) { return; } String fileName ((FileSplit) context.getInputSplit()).getPath().getName(); String[] words line.trim().split(\\s); for (String word : words) { outKey.set(word fileName); outValue.set(1); context.write(outKey, outValue); } } }在Driver里可以指定一个Combiner作用是在Map端先把相同“单词文件名”的计数加起来减少网络传输量。Combiner是MapReduce一个非常重要的优化手段但必须保证它的逻辑是Reduce逻辑的子集否则计算结果就会出错。对于倒排索引这种“先聚合词频”的场景Combiner安全且高效。Reduce端拆开key汇总成“文件名:次数”列表合并到一个Text对象里输出。这种“Map端拼接、Reduce端拆解”的思路在整个MapReduce开发中非常常见请务必掌握。4. 综合实训招聘数据清洗的完整思路4.1 先看数据脏数据通常脏在哪里“实验4 MapReduce综合应用案例——招聘数据清洗”这类任务输出结果前必须先理解输入数据。招聘数据常见的脏情况我盘点一下字段缺失某些行只有职位没有公司名、城市或薪资split之后数组长度不足。薪资格式不统一同样是月薪“10k-20k”、“1w-1.5w”、“面议”、“10K-15K”五花八门。城市写法混乱同一个城市有“北京”、“北京市”、“Beijing”多种写法。重复记录同一条招聘信息在数据里出现多次。字段内部嵌逗号公司地址里带逗号直接把整行按逗号拆分会错位。发布时间空值或乱码。拿到数据的第一步不是写代码而是先在HDFS上看一眼原始样本字段用什么分隔符一共多少列每列大概是什么含义。实训平台往往会在题目描述里给样例但样例只展示开头几行真实数据里那些异常行才是考察点。我会用hdfs dfs -cat 输入路径 | head -100这样的命令抽样观察确认分隔符和字段数量。4.2 清洗与分析的MapReduce设计清洗任务我一般设计成两轮MapReduce。第一轮负责清洗和去重第二轮负责统计分析。第一轮的Map只做一件事逐行解析、校验、过滤。字段数量不足的行直接丢弃薪资解析失败的置为null城市名通过字典映射统一。每丢弃一条用计数器记录一次方便在日志里看到清洗量context.getCounter(CleanStats, bad_lines).increment(1); context.getCounter(CleanStats, salary_null).increment(1);解析薪资的函数注意单位换算。这里我写一个简化的版本private static double parseSalary(String raw) { if (raw null || raw.trim().isEmpty()) { return -1; } raw raw.trim().toLowerCase(); if (raw.contains(面议)) { return -1; } double unit 1; if (raw.contains(k)) { unit 1000; raw raw.replace(k, ); } else if (raw.contains(w)) { unit 10000; raw raw.replace(w, ); } if (raw.contains(-)) { String[] parts raw.split(-); if (parts.length ! 2) { return -1; } try { double min Double.parseDouble(parts[0].trim()); double max Double.parseDouble(parts[1].trim()); return (min max) / 2 * unit; } catch (NumberFormatException e) { return -1; } } try { return Double.parseDouble(raw) * unit; } catch (NumberFormatException e) { return -1; } }解析完字段后将有效字段拼接成一行文本输出key可以用去重标记比如“公司名职位名发布时间”的拼接value随意给一个占位符。Reduce端负责去重同一个key只取第一条输出。第二轮的统计分析逻辑就比较灵活了。如果题目要求“各城市平均薪资Top10”Map端以城市为key、薪资为valueReduce端累加并求平均如果想进一步输出Top10还需要再做一次自定义排序。所以排序和分组的知识点不是孤立的它们在综合实训里会组合出现。综合实训做会了一类题所有“清洗统计”的套路基本都能迁移第一步探查数据第二步清洗过滤第三步去重第四步按业务字段做统计排序。4.3 实训平台做题的五个实用习惯刷实训关卡时建议养成这几个习惯能节省大量时间第一先读题目给出的输入和输出格式再动手写代码。实训平台经常会在代码区预留TODO只需要补全特定的类或方法不要上来就改主类的类名否则平台可能找不到入口。平台判题靠的是指定类名改类名等于直接0分。第二本地用Local模式验证后再提交。Local模式设置很简单把mapreduce.framework.name设为localfs.defaultFS设为file:///输入输出都改成本地路径也能正常跑通整个流程只是没有分布式效果。第三注意处理空行和首尾空格。文本文件最后一行后面往往有个换行符split后可能出现空字符串不做判空处理数组越界的分分钟。第四代码里写好计数器。清洗类任务用计数器统计过滤了多少行“脏数据”跑完作业后看日志里的数值就能判断逻辑是否生效。如果投稿0条有效数据先怀疑是不是所有行都被误判成脏数据过滤掉了。第五不要同时修改框架自带的方法。比如WritableComparator的继承写法在多个关卡中是一致的改一个地方可能导致其他关卡失效。5. 运行与调试这些问题我几乎每次都遇到5.1 常见异常速查表我整理了实训期间出现频率最高的几类问题做成速查表遇到对应关键字直接查报错或现象常见原因处理建议java.lang.ClassCastExceptionMap输出类型与Reduce输入类型不一致检查Driver里setMapOutputKeyClass和ValueClassjava.lang.reflect.InvocationTargetException自定义Bean没有无参构造器补一个public无参构造器java.io.EOFExceptionreadFields和write的字段顺序不一致调整字段读写顺序一一对应PathExistsException: Output directory already exists输出目录已经存在删除原目录或换一个新的输出路径java.io.FileNotFoundException: Input path does not exist集群上输入路径写错先hdfs dfs -ls确认路径存在Container killed on request容器内存不够调大mapreduce.map.memory.mb、mapreduce.reduce.memory.mbTextSerializationException尝试用JobConf直接序列化自定义类确认所有自定义类都实现了Writable接口运行成功但输出结果为空过滤条件过严或输入路径为空用计数器查看读取了多少行、过滤了多少行其中“输出目录已存在”几乎每个阶段都会踩到Hadoop不会自动覆盖输出目录第二次运行时如果不改路径或先删除旧目录任务直接失败。实训平台上很多同学连续提交两次作业就会卡在这一步。5.2 数据倾斜Map端加盐 二段聚合数据倾斜属于进阶问题但招聘数据清洗这类综合案例里很容易出现北京、上海这些城市的岗位量远大于其他城市按城市统计时处理北京的Reduce任务可能跑很久其他Reduce早就结束了整体任务时间被一个长尾拖死。常规解法叫“加盐”。第一轮Map时把key从“城市”改成“城市随机数”比如“北京_1”、“北京_5”这样同一个城市的记录被随机拆到多个Reduce上先做一次局部聚合。第一轮输出的中间结果是“城市 —— 部分平均值”。第二轮再按城市聚合把多个局部结果合起来得到最终均值。加盐必须注意第一阶段不要做全局排序。加了随机后缀的key排序结果和业务排序完全不对应排序语义会被破坏。所以排序类的需求不适合直接加盐需要先聚合、再排序两个阶段分开来实现。这个经验写到面试里是很加分的亮点。5.3 本地调试三板斧调试MapReduce和调试普通Java程序不一样你没法直接在IDE里全程断点跟到集群上。我常用的三板斧第一日志打印。在map或reduce方法里写System.err.println(processed: line)输出会进container日志跑完job以后在应用日志里能看到。虽然log4j更规范但在实训环境里System.err是最快、最直接的。第二计数器。前面已经提了几次它是排查脏数据过滤情况的利器。写完清洗逻辑之后加几个Counter跑完以后看一眼数值是否合理比反复看输出文件高效得多。第三小数据量验证。抽取测试数据集时不要取那种清清楚楚的干净样例而是故意留几个脏数据行看看自己的过滤逻辑能不能正确处理。真机上傻跑几个GB数据一次提交要几分钟甚至更久全用来试错太浪费。尽量在本地用几千行数据把逻辑调稳再上集群跑全量。在我个人带实训的经验里大家卡住的地方大多不是写不出map和reduce而是搞混了shuffle阶段“排序、分区、分组”这三件事各自的职责。默认排序帮不了太多忙绝大多数场景都要靠自定义Bean和自定义比较器来实现业务逻辑所以这篇内容里我刻意把排序、分组和倒排索引写得很细因为这些点一旦想通后面写综合实训就会顺很多。拿到一道新题先问自己三个问题输入长什么样输出长什么样中间哪些值需要聚合或排序。把这三个答案写在纸上再动手比直接打开IDE死磕代码高效得多。