用MapReduce实现TF-IDF关键词提取：原理与工程实战

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）：某个词在文档中的出现次数除以该文档的总词数
• 逆文档频率（IDF）：log（文档总数 / （包含该词的文档数 + 1））
• TF-IDF = TF × IDF

这个“+1”是为了防止分母为零，日志底数用10还是用自然对数影响不大，因为我们只关注相对大小。

如果一次MapReduce就想把上面三行公式全算出来，我劝你别这么想。不是做不到，是代码复杂度会急剧膨胀，而且每个阶段都要Shuffle排序，中间结果落地超大，实际跑起来慢得没法看。最合理的做法，是把TF-IDF拆成三轮MapReduce，让每一轮只完成一件清晰的事情。

第一轮统计每个文档里每个词的出现次数，输出词频（TF）。这里的难点在于还需要同时得到每篇文档的总词数，用来做标准化。第二轮统计每个词出现在多少篇文档里，输出文档频率（DF），同时拿到文档总数，为计算IDF做准备。第三轮把前面两部分数据Join起来，最终计算每个词在每篇文档中的TF-IDF值并输出。

2.2 三个Phase的输入输出设计

我把三轮的数据流画在脑子里，大概是这样：

很多教材喜欢把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命令上传到HDFS：

hdfs dfs -mkdir -p /tfidf/input hdfs dfs -put /usr/local/data/tfidf/input/*.txt /tfidf/input/ hdfs dfs -cat /tfidf/input/编程.txt | head -50

3.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.java

Phase3不需要自定义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 Mapper<LongWritable, 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阶段一次性输出所有结果，同时输出文档总词数。具体做法是自定义一个Partitioner：

public class DocPartitioner extends Partitioner<Text, 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就可以这样实现：在内存中用一个Map<String, Integer>记录每个词的出现次数，再遍历完所有词之后统一输出。由于同一文档的词全在一个Reducer内，我们可以顺便算出文档总词数：

public class WordCountReducer extends Reducer<Text, Text, Text, Text> { @Override protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { String docName = key.toString().split("\t")[0]; Map<String, 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.Entry<String, 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-IDF

5.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 Reducer<Text, Text, Text, Text> { @Override protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { Set<String> 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 { Set<String> 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 Mapper<LongWritable, Text, Text, Text> { private Map<String, 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三个文档里，DF=3。

假设总文档数量为4，那么IDF为：ln(4 / (3 + 1)) = 0。第三轮算出的TF-IDF就是0，说明该词在4篇文档中出现了3篇，区分度很差。

再拿一个稀有词“贝叶斯”，如果它只出现在文档A里，DF=1，IDF=ln(4 / 2)≈0.693，TF值如果是15/1200=0.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 代码结构小复习

方便你对照复盘，我把三轮的核心代码职责再列一遍：

• Phase1：Map阶段输出“文档名+词”的Key和计数，Partitioner按文档名分区，Reduce端计算词频和文档总词数
• Phase2：Map按词分组，Reduce按文档名去重计数，输出DF
• Phase3：DistributedCache加载IDF，Map端读出词频并直接计算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整个计算模型的理解绝对会上一个台阶，远远超出了只看不练的效果。