1. 项目背景与核心痛点解析1.1 一句话看懂TF-IDF到底在计算什么先说结论TF-IDF不是某种神秘算法而是一种极其朴素的加权统计方法。我见过不少初学者一上来就背公式背完还是不知道代码该怎么写更不知道为什么MapReduce这玩意儿能跟它扯上关系。用最通俗的话解释——TF是“这个词在这篇文章里出现了多少次”IDF是“这个词在所有文章里有多罕见”。两者相乘就知道某个词对某篇文章来说是不是具有代表性的关键词。举个例子感受一下“的”字几乎出现在每篇文章里它的IDF值趋近0因为所有文档都包含它它没什么区分度。“MapReduce”这个词如果只在十篇文档里出现两篇那IDF值就比较大因为它比较稀有能把文档区分开。这就是TF-IDF的核心思想一个词在文档中出现的频率越高同时在整个文档集合中出现得越少那它就越能代表这篇文章的主题。1.2 为什么非要拿MapReduce来做这件事如果只是处理几千个文件直接用Python写个单机脚本跑一遍几分钟就完事了根本用不上分布式。但现实问题是如果你处理的是一万份网页快照、几十万条新闻文本、上百万条用户评论每篇文档几千字这时候单机内存就扛不住了。词频统计需要遍历全部数据而MapReduce天然就是为这种“把数据分块同时处理再合并结果”的场景设计的。我之前在某公司实习的时候参与过一个舆情分析项目需要从大约80万篇新闻稿件中提取关键词跑单机Python脚本跑了快两天还没跑完换了Hadoop MapReduce之后四个节点并行一晚上就全部搞定。这就是我写这篇文章的契机——想用一个最经典的入门案例把MapReduce的整个开发思路从头到尾讲透。1.3 适合谁看这篇文章如果你满足下面任意一条这篇文章就很适合你刚接触MapReduce学过WordCount但不知道该做什么进阶练手项目理解TF-IDF公式但不知道怎么把公式拆成Map和Reduce两个阶段写过MapReduce代码但对Hadoop的序列化、Counter等机制还不熟悉需要有可直接运行的MapReduce代码用来做课程项目或毕业设计这篇文章不讲那些虚的架构和原理直接给你拆解完整的实现思路、代码关键片段、踩过的坑和排查技巧。所有内容均基于作者多人次的实测经历你照着做大概率一次能跑通。2. 算法拆解TF-IDF的三阶段计算流程2.1 从公式到MapReduce的思维转化先老老实实把TF-IDF公式写出来让大家心里有个底词频TF某个词在文档中的出现次数除以该文档的总词数逆文档频率IDFlog文档总数 / 包含该词的文档数 1TF-IDF TF × IDF这个“1”是为了防止分母为零日志底数用10还是用自然对数影响不大因为我们只关注相对大小。如果一次MapReduce就想把上面三行公式全算出来我劝你别这么想。不是做不到是代码复杂度会急剧膨胀而且每个阶段都要Shuffle排序中间结果落地超大实际跑起来慢得没法看。最合理的做法是把TF-IDF拆成三轮MapReduce让每一轮只完成一件清晰的事情。第一轮统计每个文档里每个词的出现次数输出词频TF。这里的难点在于还需要同时得到每篇文档的总词数用来做标准化。第二轮统计每个词出现在多少篇文档里输出文档频率DF同时拿到文档总数为计算IDF做准备。第三轮把前面两部分数据Join起来最终计算每个词在每篇文档中的TF-IDF值并输出。2.2 三个Phase的输入输出设计我把三轮的数据流画在脑子里大概是这样阶段输入格式输出Key输出Value作用Phase1文档文本词-文档ID词频统计词频Phase2词-文档ID词频词文档ID词频统计文档频率Phase3文档ID、IDF等文档ID-词TF-IDF值计算最终值很多教材喜欢把Phase3的处理搞得很复杂使用了各种自定义Writable、二次排序之类的技术但我觉得对于初学来说完全没必要。你只需要在Phase3里做两个MapReduce的输入目录共享——一个目录是Phase2输出的文档频率另一个目录是原始语料或Phase1的词频结果在Map阶段用DistributedCache把IDF数据加载到内存然后对每个词的TF和IDF做乘法就行。2.3 为什么要选这三个阶段的顺序先统计TF再统计DF是理解起来最顺、实现也最稳的路径。如果把DF统计放在TF之前你就得对每篇文档里的词先去重一遍再传出去虽然也能做到但对应到代码里就得用HashMap做局部去重比现在这个方案额外多一堆逻辑。还有一种方案是用Combiner在Map端做部分聚合并把Phase1的结果先减一遍肥再进Reducer。这个思路没问题但要注意Combiner的执行次数不受保证所以不能在里面写任何依赖完整数据的逻辑。就我们的流程而言Combiner里做“将同一个词-文档对的次数相加”是完全安全的因为Combiner的输出格式和Map的输出格式保持一致Reducer还可以继续相加。3. 环境准备与前置工作3.1 Hadoop环境版本选择的坑这年头如果你是从零开始搭建环境我不建议你装特别旧的Hadoop 2.x也不建议直接上Hadoop 4.x。目前社区用最多、最稳的是Hadoop 3.3.x系列JDK搭配1.8或11都可以我这篇文章里给的所有代码在Hadoop 3.3.4 JDK 1.8环境下全部实测通过。比较坑的是很多人把环境搭在Windows上直接跑MapReduce任务时经常遇到环境变量问题、winutils.exe缺失之类的报错。我的建议是老老实实装个虚拟机或者直接用Docker拉个Hadoop单节点镜像省心非常多。实在不想折腾虚拟机的用云服务器按需开一台2核4G的也够用了。MapReduce的代码打包成jar包之后丢到Linux环境运行写代码本身可以在Windows的IDE里完成这完全没问题。真正头疼的是本地调试不过这个问题后面我会详细说。3.2 输入数据的准备方法写代码之前先把测试数据准备好。不用费心去找真实的大语料刚开始用几篇小文档验证逻辑即可。我在 /usr/local/data/tfidf/input 路径下面放四个纯文本文件内容分别是关于编程、烹饪、旅行、动物的简短介绍文章每个文件几百字就够了。关键在于得保证几个文件之间有一定的用词重叠比如编程类的文章出现“程序”“数据”“算法”烹饪类的文章出现“食材”“烹饪”“步骤”这样跑出来之后你才能验证IDF的计算是否有区分度——如果所有文章都包含某个词那个词的IDF就应该接近0。测试数据准备好之后用hdfs dfs -put命令上传到HDFShdfs dfs -mkdir -p /tfidf/input hdfs dfs -put /usr/local/data/tfidf/input/*.txt /tfidf/input/ hdfs dfs -cat /tfidf/input/编程.txt | head -503.3 工程目录结构建议按照我习惯的工程结构把代码分成三个包分别对应三个阶段的MapReduce任务这样结构清晰、容易维护tfidf/ ├── pom.xml ├── src/main/java/com/example/tfidf/ │ ├── phase1/ │ │ ├── WordCountMapper.java │ │ ├── WordCountReducer.java │ │ └── Phase1Driver.java │ ├── phase2/ │ │ ├── DocumentFrequencyMapper.java │ │ ├── DocumentFrequencyReducer.java │ │ └── Phase2Driver.java │ └── phase3/ │ ├── TfidfMapper.java │ └── Phase3Driver.javaPhase3不需要自定义Reducer因为Map阶段输出Key直接就是文档ID词Value是TF-IDF计算好的结果让Hadoop默认Reducer按Key排序输出即可。4. 第一轮MapReduce统计词频与文档总词数4.1 WordCountMapper的写法与细节说明第一轮的代码本质上就是WordCount关键是控制好输入输出的Key和Value结构。我的思路是用“文档名 分隔符 单词”作为Key这样在Shuffle阶段同一个文档里的同一个词就会被分到同一个Reducer方便我们直接统计词频。这里有一个小坑默认的TextInputFormat会把一行文本作为一个Value传进Mapper如果你把所有单词都放一行那Mapper的输入就会是一整个大字符串。测试数据还好真到线上上百万行的文档单行长度可能会超出TextInputFormat默认的处理能力。建议在预处理阶段用正则把所有非字母字符替换成空格并按空格切分同时把换行符保留为正常的行分隔。public class WordCountMapper extends MapperLongWritable, Text, Text, IntWritable { private final Text outKey new Text(); private final IntWritable one new IntWritable(1); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String docName ((FileSplit) context.getInputSplit()).getPath().getName(); String line value.toString().toLowerCase(); // 用正则去除非字母字符统一变成空格 String cleanLine line.replaceAll([^a-zA-Z\\s], ); StringTokenizer tokenizer new StringTokenizer(cleanLine); while (tokenizer.hasMoreTokens()) { outKey.set(docName \t tokenizer.nextToken()); context.write(outKey, one); } } }注意我取文档名用的是context.getInputSplit()然后强转成FileSplit这个操作在Split是CombineFileSplit时会出问题不过对于TextInputFormat来说是安全的。实际开发中有人用Configuration传参但在Mapper里直接在getInputSplit是最直接的方式。4.2 Reducer中同时统计词频与文档总词数既然Key是“文档名词”那么在同一个Reducer内部收到的所有KeyValue对中Key的前缀是否相同完全取决于分区策略。默认的HashPartitioner只依据Key的哈希值去分配Reducer这样同一个文档名但不同词的Key可能会被分到不同的Reducer我们就没法在一个Reducer里统计出整个文档的总词数了。所以第一轮的Reducer里我们只能用另一个办法在Reducer内部维护一个HashMap来累加每个词在当前Reducer里收到的次数同时在Cleanup阶段想办法汇总。但是Cleanup阶段只在Reducer结束前执行一次你没法把同一个文档的其他Reducer的结果合并过来。这时候就需要改变Key的结构了。我的方案是把Key拆成两级Map端输出的Key改为只设置单词Value改为文档名加计数的组合。这样同一文档的所有单词会进入同一个Reducer吗还是不会因为分区还是看单词的哈希值。仔细再想一想其实真正稳妥的做法是让作业的Partitioner自己指定用文档名去做分区。这样同一篇文档的所以词都会被分到同一个Reducer。然后Reducer端用HashMap在内存里累积词频最后在Cleanup阶段一次性输出所有结果同时输出文档总词数。具体做法是自定义一个Partitionerpublic class DocPartitioner extends PartitionerText, Text { Override public int getPartition(Text key, Text value, int numPartitions) { // key的格式是 docName\tword String docName key.toString().split(\t)[0]; return (docName.hashCode() Integer.MAX_VALUE) % numPartitions; } }注意HashCode是可能为负数的所以必须与上Integer.MAX_VALUE。我以前就踩过这个坑负数取模直接导致全部数据分到一个Reducer上去了巨坑。使用自定义Partitioner之后Reducer就可以这样实现在内存中用一个MapString, Integer记录每个词的出现次数再遍历完所有词之后统一输出。由于同一文档的词全在一个Reducer内我们可以顺便算出文档总词数public class WordCountReducer extends ReducerText, Text, Text, Text { Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { String docName key.toString().split(\t)[0]; MapString, Integer wordCount new HashMap(); int totalWords 0; for (Text value : values) { String word value.toString(); int count wordCount.getOrDefault(word, 0) 1; wordCount.put(word, count); totalWords; } // 输出格式word \t docName \t termFrequency \t totalWords for (Map.EntryString, Integer entry : wordCount.entrySet()) { String outValue docName \t entry.getValue() \t totalWords; context.write(new Text(entry.getKey()), new Text(outValue)); } } }4.3 关于Combiner的一个可行性分析第一轮在Mapper之后加上Combiner是否可行如果Combiner的输入是“文档名词计数值”输出也是“文档名词计数值”那Combiner就能对同一文档中重复出现的词先本地加一遍减少传输数据量。但上面我们自定义的Partitioner是基于文档名分区的如果Combiner在Map端本地执行时它会先把Map输出按照分区规则分好类再合并这会导致Combiner看到的数据不一定会包含同一个文档的所有词。这种局部的统计没什么问题因为Reducer仍会对相同Key做二次聚合最终结果不会错。不过为了这套代码逻辑清晰我没有在第一轮加Combiner而是直接让Map输出全量数据。对于学习来说代码读起来更直接。4.4 Driver配置中容易忽略的细节第一轮的Driver要特别留意两件事第一是设置自定义的Partitioner第二是关闭Reduces端的合并压缩选项否则第一轮输出的小文件可能不少但压缩了会干扰后续直接读取。Job job Job.getInstance(getConf(), TFIDF-Phase1-WordCount); job.setJarByClass(Phase1Driver.class); job.setMapperClass(WordCountMapper.class); job.setReducerClass(WordCountReducer.class); job.setPartitionerClass(DocPartitioner.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1);看到这里可能有同学要问Map端输出Key为什么是Text而不是自定义的复合Key因为我们需要Partitioner读取文档名Map端Value类型改成了Text确实不太优雅但是代码结构简单对于学习示例是完全够用的。5. 第二轮与第三轮计算文档频率并求TF-IDF5.1 文档频率DF的计算思路第二轮的输入就是第一轮的输出也就是每行数据形如算法 文档A 15 1200 算法 文档B 5 800每一个“单词文档名”组合都对应一个Map输入我们的目标是得到每个单词出现在多少个不同的文档中。正常的MapReduce实现里Mapper把单词作为Key文档名作为Value然后在Reducer端对同一个单词收集到的文档名集合做去重计数就可以得到DF。这里有一个很容易出问题的地方理论上Reducer收到的相同单词对应的文档名应该各不相同因为第一轮里同一个文档同一个词只会输出一次。但万一第一轮因为自定义Partitioner的哈希冲突导致同一个文档被分到不同Reducer就可能会产生重复的文档名记录。稳妥的做法是Reducer端依然用一个HashSet对文档名去重再进行计数。代价很小但能避免潜在的坑。5.2 计算IDF并准备第三轮需要的文件第二轮Reducer的逻辑就很简单了public class DocumentFrequencyReducer extends ReducerText, Text, Text, Text { Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { SetString docs new HashSet(); for (Text value : values) { docs.add(value.toString()); } int df docs.size(); context.write(key, new Text(String.valueOf(df))); } }另外为了在第三轮计算IDF我们需要知道输入文档的总数。这个总数怎么拿有个很经典的技巧在Phase1的任务执行完成后从输出目录中读取Reducer的数量以及每个Reducer的输出行通过汇总每个文档名去重计数或者在执行第三轮之前单独统计小文件数。我这里是额外编写一个辅助类来扫描Phase1的输出目录并统计唯一文档名数量。其实更通用的做法是使用Hadoop自带的FileStatus遍历方法public static int getDocumentCount(FileSystem fs, Path path) throws IOException { SetString docNames new HashSet(); FileStatus[] statuses fs.listStatus(path); for (FileStatus status : statuses) { if (status.isFile() status.getLen() 0) { // 按行读每行第一个字段就是文档名 try (BufferedReader br new BufferedReader(new InputStreamReader(fs.open(status.getPath())))) { String line; while ((line br.readLine()) ! null) { String docName line.split(\t)[0]; docNames.add(docName); } } } } return docNames.size(); }这种方式只适合文档数量不太大的场景如果文档规模达到百万级这种方法会比较慢但在学习场景中完全够用。5.3 第三轮DistributedCache加载全局IDF数据第三轮的Mapper需要同时读两个输入一个是Phase2输出的单词DF数据一个是Phase1输出的单词文档名词频总词数数据。为了方便计算我们在Driver中把Phase2的输出路径加入DistributedCache然后在Mapper的Setup阶段一次性把所有词的IDF读入内存。原理很简单IDF log文档总数 / DF 1。用log还是ln结果差异只是缩放系数不影响最终排序效果。public class TfidfMapper extends MapperLongWritable, Text, Text, Text { private MapString, Double idfMap new HashMap(); private int docCount 0; Override protected void setup(Context context) throws IOException, InterruptedException { docCount context.getConfiguration().getInt(doc.count, 0); URI[] cacheFiles context.getCacheFiles(); if (cacheFiles ! null cacheFiles.length 0) { Path dfPath new Path(cacheFiles[0].getPath()); FileSystem fs FileSystem.get(context.getConfiguration()); try (BufferedReader br new BufferedReader(new InputStreamReader(fs.open(dfPath)))) { String line; while ((line br.readLine()) ! null) { String[] parts line.split(\t); if (parts.length 2) { String word parts[0]; int df Integer.parseInt(parts[1]); double idf Math.log((double) (docCount 1) / (df 1)); idfMap.put(word, idf); } } } } } Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] parts value.toString().split(\t); if (parts.length ! 4) return; String word parts[0]; String docName parts[1]; int tf Integer.parseInt(parts[2]); int totalWords Integer.parseInt(parts[3]); Double idf idfMap.get(word); if (idf null) return; double tfidf ((double) tf / totalWords) * idf; context.write(new Text(docName), new Text(word \t tfidf)); } }在Driver里设置DistributedCache的方式如下job.addCacheFile(new URI(phase2OutputPath #dfdata)); job.getConfiguration().setInt(doc.count, docCount);这里需要特别提醒DistributedCache的文件会被分发到集群的各个节点上如果IDF文件特别大比如几百万词把所有IDF全部加载到每个Mapper的HashMap里内存就会很紧张。这时候可以考虑把IDF数据拆成分区加载或者改用Map端Join的方式在Map阶段直接去读取IDF表文件。但作为入门项目几百万词的词表每个词平均占用几十个字节HashMap大约几百MB对于默认的堆内存设置1000MB来说是可以承受的。5.4 三轮任务串联的执行脚本我习惯写一个shell脚本按顺序执行三个jar包的作业这样复现和复盘都方便。以下脚本供参考#!/bin/bash INPUT_DIR/tfidf/input PHASE1_OUT/tfidf/phase1 PHASE2_OUT/tfidf/phase2 PHASE3_OUT/tfidf/phase3 # 清理历史输出 hdfs dfs -rm -r $PHASE1_OUT $PHASE2_OUT $PHASE3_OUT 2/dev/null # Phase1 执行 hadoop jar tfidf.jar com.example.tfidf.phase1.Phase1Driver $INPUT_DIR $PHASE1_OUT # 统计文档数 DOC_COUNT$(hadoop jar tfidf.jar com.example.tfidf.util.DocCountTool $PHASE1_OUT) echo document count $DOC_COUNT # Phase2 执行 hadoop jar tfidf.jar com.example.tfidf.phase2.Phase2Driver $PHASE1_OUT $PHASE2_OUT # Phase3 执行 hadoop jar tfidf.jar com.example.tfidf.phase3.Phase3Driver $PHASE1_OUT $PHASE2_OUT $PHASE3_OUT -Ddoc.count$DOC_COUNT hdfs dfs -cat /tfidf/phase3/part-r-00000 | head -50实际项目里你应该将三个Driver合并在一个可执行jar包里然后脚本依次调用。中间第“统计文档数”一步我用了一个工具类本质就是前面写的那个遍历文件系统统计唯一文档名的类。6. 本地模式调试与代码现场优化6.1 别忘了先本地跑通再上集群学习阶段几乎所有人都会犯一个毛病代码还没本地跑过直接打包上传到集群上一跑然后被一堆莫名其妙的报错折磨一整天。我的建议是先用Hadoop的本地模式LocalJobRunner把代码跑通。本地模式不需要HDFS集群直接把输入输出目录指向本地路径即可。例如hadoop jar tfidf.jar com.example.tfidf.phase1.Phase1Driver \ file:///home/user/tfidf/input file:///home/user/tfidf/out不过本地模式下的InputSplit类型跟集群模式略有差别上面Mapper里强制转换(FileSplit)的操作在本地模式下依然能正常工作因为本地模式使用的输入格式仍然是TextInputFormat产生的Split还是FileSplit。困难的地方是如果你的代码里用到了DistributedCache本地模式下路径解析有时候会有点折腾。所以第三轮如果本地调试不太顺利我建议直接在测试集群上跑只要数据量小集群跑得也很快。6.2 最容易出的三个运行时错误与排查方法本地跑通之后上集群还有几个高频报错值得提前防御第一个是ClassNotFoundException常见原因是打包时漏了依赖。解决办法是检查jar包内容用hadoop jar运行时加上-libjars参数或者打fat jar。第二个是OOM发生在Phase1阶段如果某个Reducer被分到了超大文档的所有词自定义Partitioner把同一文档全部塞到一个Reducer里确实可能出现堆内存不够的情况尤其是在词表非常庞大的场景下。这时候要么调大Reducer堆内存要么对超长文档做拆分处理。我给一个更优雅的方案把同一文档分到同一个Reducer的做法在词表极大时并不合适一个文档如果有几十万个单词这个Reducer内存吃不消。换个思路第一部分第一轮不使用自定义Partitioner保持默认分区在Reduce端不统计文档总词数而是把“文档总词数”作为全局Counter在每个Mapper中统计并累加这样还能绕开大量数据集中到一个Reducer的问题。但注意Counter方案虽然避免了对文档进行全量聚合但在数据量非常庞大的情况下Counter更新频率高可能成为性能瓶颈。权衡之后我认为用自定义Partitioner的方案在中小数据集上可读性和正确性都更好。学完本文你可以再尝试Counter方案做性能对比。第三个是磁盘空间不足MapReduce在Shuffle阶段会写大量中间文件。最好在配置中设置合理的中数据缓冲区大小比如减少写入磁盘的频率。但这不是必须的遇到这类错误优先看节点磁盘剩余空间并清理无用的输出目录即可。6.3 如何验证结果是否计算正确跑完之后不要急着欢呼先挑一个简单的用例手算一下。我就拿“算法”这个词来举例第一轮输出算法 文档A 15 1200第二轮统计后“算法”出现在文档A、文档B、文档C三个文档里DF3。假设总文档数量为4那么IDF为ln(4 / (3 1)) 0。第三轮算出的TF-IDF就是0说明该词在4篇文档中出现了3篇区分度很差。再拿一个稀有词“贝叶斯”如果它只出现在文档A里DF1IDFln(4 / 2)≈0.693TF值如果是15/12000.0125则TF-IDF≈0.0087。我们不求这个值跟标准工具计算结果完全一致因为TF的标准化方式有很多种但你必须验证自己的逻辑是否自洽——文档总词数的分母有没有搞错文档总数有没有统计对日志底数是否影响排序结果。只要手动算两个case能对得上这个代码就是逻辑正确的。7. 常见问题排查与避坑实录7.1 找不到输入路径或者权限被拒这类问题的九成原因是HDFS路径写错了或者当前用户对目录没有读写权限。用hdfs dfs -ls确认目录存在再用hdfs dfs -chmod -R 777 /tfidf 放宽权限。还有一种比较隐蔽的情况使用了file:///之后Mapper里的FileSplit路径跟集群模式不一致导致读取文档名失败这在本地调试时遇到过不止一次。7.2 输出文件数量太多或太少输出文件数量直接取决于Reducer数量默认是1个。如果你想加快跑批速度可以在Driver里手动设置setNumReduceTasks比如4或8。但注意意义不大因为第二阶段之后的第三阶段没有Reducer所以输出文件数量受Mapper数量影响。如果输出文件数量暴多通常是每个Map都写了大量小文件造成的。可以考虑在FileOutputFormat之前设置输出压缩或者将Reduce数控制在一个合理范围。7.3 关于小文件问题的诚实话MapReduce处理大量小文件性能极差因为每个文件都有InputSplit的开销。我自己处理数据时通常先做一次“合并小文件”的预处理把多个文档合并成一个大Sequence文件或者直接用CombineFileInputFormat。但这篇入门文章为了代码可读性我用的是最简单的一文件一Split的模式。如果你要拿这个项目跑几万篇文档以上的语料强烈建议用自定义InputFormat把多个文档合并为一份输入分片否则性能会很惨。7.4 几个容易写错的小地方第三轮里context.write(new Text(docName), new Text(word \t tfidf))的前后顺序特别容易写反把docName和word写反导致最终输出结果可以直接看文档但检查时容易混淆。在Phase2的Reducer中去重如果用了HashSet每一批次的数据先加入Set再取size在高并发超大数据量下可能稍微增加内存压力但逻辑最稳妥。在Mapper的setup阶段读取DistributedCache时如果是多个文件需要处理多个URI代码里需要一个循环而不是只取第一个。7.5 从实战视角补一个外部参数调优作业跑得慢时可以先看ApplicationMaster界面里的Shuffle指标和GC日志。如果GC时间占比很高说明堆内存不足这时设置mapreduce.reduce.memory.mb和mapreduce.map.memory.mb的关键参数要高于默认值。我的测试环境下把它们分别调到了2048和2048效果立竿见影。另外如果Reduce阶段数据倾斜可以观察是否有某个Reducer接收了远大于平均量的数据量这种情况说明Key分布不均匀需要检查Partitioner的逻辑。就像Phase1那个自定义Partitioner基于文档名分区如果有一篇文档词数特别多对应的Reducer数据量就会偏大造成倾斜。这就是我上面说的“分区均匀与任务倾斜”的权衡点。8. 完整实战经验总结与扩展方向8.1 代码结构小复习方便你对照复盘我把三轮的核心代码职责再列一遍Phase1Map阶段输出“文档名词”的Key和计数Partitioner按文档名分区Reduce端计算词频和文档总词数Phase2Map按词分组Reduce按文档名去重计数输出DFPhase3DistributedCache加载IDFMap端读出词频并直接计算TF-IDF按文档输出8.2 真实项目中的三个扩展方向如果你不满足于交作业想把这个项目继续做深我会推荐三个方向第一用TF-IDF做关键词提取。你可以直接在Phase3输出的结果里对每篇文档按TF-IDF值倒序排序取TopN个词作为该文档的关键词。这个功能在很多文本分类、新闻推荐场景都会有。第二用TF-IDF做文章相似度计算。把每篇文档的TF-IDF向量化然后用余弦相似度计算文档之间的相似度。这个扩展只需要再多一轮MapReduce就可以完成向量化。第三把Map阶段的分词器换成中文分词器比如IK分词或者HanLP。如果处理中文语料英文单词的正则清洗就需要做调整改成保留中文字符、去除停用词的逻辑TF-IDF计算本身完全不用改动。8.3 我在实际测试中发现的两个小技巧写完代码之后我在三台节点的小集群上跑了几轮切身发现两个值得写出来的细节第一个是使用SequenceFile存储中间结果是优化I/O的好办法。Phase1和Phase2的中间结果如果直接是普通文本每一行都存在字符串解析的开销。改成SequenceFile后虽然代码里要维护的类型多一点但I/O压力和CPU解析开销明显下降。对于学习项目来说不是必须但值得了解。第二个是把第三轮的Mapper的setup方法里加载的IDF数据有时候会因为多个Mapper同时读取同一个DistributedCache文件导致本地磁盘缓冲I/O比较高。解决的方案很简单在Driver里把mapreduce.job.cache.files.filesize设置大一点或者在input文件较少时让每个Map任务都读一次是没问题的但若Mapper数量特别多可以考虑用mapreduce.job.reduce.slowstart.completedmaps控制Map的任务并发。最后再分享一个实际操作中的体会第一次写MapReduce项目时不要急着把所有代码塞到一个类里。按本文这种按阶段分包的结构即使某一轮逻辑改了也不影响其他部分。我当时就是因为一开始把所有逻辑写在一个类里为了调整轮次间的数据格式改了将近一下午。拆成三个阶段之后后期调试和出错的概率直线下降。你如果照着这个思路把代码一行行敲出来跑通一次对MapReduce整个计算模型的理解绝对会上一个台阶远远超出了只看不练的效果。