简介这份Hadoop词频统计完整版是一份基于Hadoop2.2.0的MapReduce项目资源面向正在学习Hadoop分布式计算、准备课程实验或入门MapReduce编程的开发者能够直观展示词频统计从文件切分、映射、洗牌到归约的完整执行链路。压缩包共17个文件包含7个Java源文件及对应编译好的7个class文件另有一个约十万单词的英文测试文本以及.project和.classpath工程配置文件整体仅154KB小巧轻便。目前已有5868人学习下载。配套的源码与编译产物可帮助读者直接导入Eclipse运行验证并通过阅读Mapper与Reducer实现理解Hadoop编程模型中键值对的处理逻辑自带的十万词级语料也能用于测试不同分词规则、自定义计数器或调整reduce数量等扩展练习是一份适合边看博客详解边动手实践的完整入门素材。1. Hadoop词频统计从“能跑通”到“完整版”差了多远Hadoop词频统计是几乎每个入门大数据的人都会见过的经典案例代码量不大逻辑也直白把一堆文本拆成单词再按词累加次数。但“完整版”这三个字才是关键——真正在集群上把这个任务跑完、跑稳、跑出可复用的结果涉及的远不止那几十行Java代码。打包、提交、参数调优、日志排查、输出设计每一步都有能让新手卡住半天的细节。我从第一次在单机伪分布式上跑通WordCount到后来在真实集群上处理上GB的日志文本最大的感受是这个项目像一把钥匙理解它MapReduce的作业生命周期、数据流和调优思路就都串起来了。这篇笔记写给两类人想从零跑通第一个Hadoop作业的新手以及想搞清楚除了默认配置还能怎么调、坑在哪里的熟手。2. 先搞懂词频统计在Hadoop里是怎么流转的角色、时机与数据流2.1 三个角色各管一段Mapper、Reducer与Driver的职责边界词频统计的MapReduce程序通常由三个类组成Mapper、Reducer和Driver有的叫Job类。很多初学者把注意力全放在Mapper和Reducer的写法上却忽略了Driver才是整个作业的“控制中枢”。Mapper负责把输入数据切成键值对在词频统计里就是每行文本拆成单词输出单词, 1这样的中间结果Reducer负责把相同单词的计数汇总输出单词, 总次数。而Driver负责配置作业的运行方式输入输出路径、哪个类是Mapper、哪个类是Reducer、设置几个Reduce任务、作业名字是什么。这里有一个常见的认知误区以为MapReduce就是“Map完接着Reduce”。实际上Mapper的输出要经过一个叫Shuffle的阶段——分区、排序、合并、归并——Reducer拿到的输入已经是按key排序并分组好的。在词频统计这个场景里Shuffle帮我们按单词把中间结果归类Reducer只需要把同一个单词下的所有计数值加起来就行。这个机制在代码里看不到但它决定了为什么Reducer的输入一定是“键值对列表”的形态。从实际工程角度看我建议新手先别急着写代码先把这三个类的关系画清楚。画清楚之后再看下面这段标准实现心里就有了一张地图哪个方法处理哪一段数据流谁在什么时候被调用。2.2 一个可编译的完整实现数据流逐段拆解先看完整的Java实现我习惯用Java写Hadoop作业因为它是Hadoop的原生语言遇到问题能搜到的资料最多。代码里有注释我们把每个关键点拆开讲。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.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; import java.util.StringTokenizer; public class WordCount { // Mapper输入 key行偏移量, value一行文本输出 单词, 1 public static class TokenizerMapper extends MapperObject, Text, Text, IntWritable { private final static IntWritable one new IntWritable(1); private Text word new Text(); Override public void map(Object key, Text value, Context context ) throws IOException, InterruptedException { // 用 StringTokenizer 按空格、制表符、换行符切分比 split() 更省内存 StringTokenizer itr new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); // 每个单词写一次 单词, 1 } } } // Reducer输入 单词, [1,1,1,...]输出 单词, 总和 public static class IntSumReducer extends ReducerText, IntWritable, Text, IntWritable { private IntWritable result new IntWritable(); Override public 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); } } // Driver组装整个作业 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); job.setCombinerClass(IntSumReducer.class); // 使用 Reducer 作为 Combiner job.setReducerClass(IntSumReducer.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); } }几点必须说明的逻辑和参数选择。Mapper里我用了StringTokenizer而不是String.split()原因在于它按分隔符逐个切分不会一次性生成整个字符串数组在处理大文件时内存占用明显更低。context.write(word, one)每遇到一个单词就输出一次单词, 1这是MapReduce的标准做法——不在Mapper里做局部累加因为Shuffle会自动帮我们按key分组。代码里设置了setCombinerClass(IntSumReducer.class)这一行在“完整版”里很关键。Combiner是运行在Map端的“迷你Reducer”它在Mapper输出后、Shuffle传输前先对同一个Map任务内部的相同单词做一次合并。比如某个Map任务里hadoop出现了10次没有Combiner就要传10条hadoop, 1给Reducer有了Combiner只传1条hadoop, 10网络传输量大幅下降。词频统计的Reducer刚好是求和逻辑满足交换律和结合律所以可以直接复用Reducer类。如果Reducer逻辑不满足这个性质就要单独写Combiner类。还有一个必须记住的点FileOutputFormat.setOutputPath传入的输出路径在作业启动时必须不存在。这是Hadoop的一种保护机制防止覆盖之前的结果。如果你重复跑同一个作业不先删掉上一次的输出目录作业会直接报FileAlreadyExistsException。我早期在这个异常上卡了好几次后面避坑章节会专门讲。2.3 兜底能力输入输出格式、键值类型与Reduce数量的默认值代码里我们没有显式设置输入输出格式但这不代表没有。Hadoop默认使用TextInputFormat它把输入文件按行切分每行作为一个valuekey是这一行在文件中的字节偏移量。所以Mapper里的key参数类型是Object——因为偏移量对算法没有意义我们只关心value的文本内容。输出格式默认是TextOutputFormatReducer输出的每个键值对被写成一行key和value之间用制表符\t分隔这正好方便后续用命令行或脚本解析。键值类型的选择也值得注意。Hadoop的Writable机制要求key和value都必须实现Writable接口才能被序列化到磁盘和网络。代码里选用了Text和IntWritable一个是UTF-8编码的文本类型一个是可变长的整型序列化类型。这两个类型的选择直接影响了Shuffle阶段的效率整数用IntWritable序列化后通常只占几个字节如果用Java自带的Integer走序列化框架字节数可能是它的数倍。关于Reduce任务数量代码里没有调用setNumReduceTasks默认值是1。这意味着所有单词的频率累加都集中在一个Reducer上完成。数据量小的时候没问题但一旦输入文本到了GB级别单个Reducer会成为瓶颈——它要拉取所有Map任务的输出处理完全部数据才算完成。完整版通常会根据集群规模和输入数据量调整这个值我会在调优章节详细展开。这里还有个容易被忽略的点如果Reduce任务数设为0作业就变成纯Map任务输出是每个Map任务各写一个文件。这可以作为一种“只做清洗不做聚合”的用法但词频统计不适合这样做因为那就没有统计了。3. 把代码跑起来环境准备、打包提交与结果验证3.1 开发环境与集群环境的差异版本匹配与依赖范围写Hadoop作业最常见的一个翻车点是本地用Maven引入的Hadoop依赖版本和集群上的Hadoop版本不一致。版本不一致不一定编译报错但运行时大概率出现奇怪的序列化异常或NoSuchMethodError。我一般会在pom.xml里显式声明与集群一致的版本号并且把依赖范围设为provided意思是编译时需要、运行时不打包进去——因为集群节点上已经有完整的Hadoop运行库了。常见做法是先确认集群的Hadoop版本再写pom。比如你所在的环境是Hadoop 3.x那么在pom里就用对应的3.x版本号如果是2.x那mapreduce相关API的包路径会有差异。Hadoop 3里的作业提交流程和2.x也有细微差别比如ApplicationMaster的启动方式但这些对WordCount透明真正影响你的是依赖版本。dependencies dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.3.6/version scopeprovided/scope /dependency /dependenciesprovided这个scope是必须的。如果你不加这个scope打包时默认会把Hadoop的所有依赖都打进jar包一个hadoop-client加上传递依赖产物动辄几十上百MB。集群提交时还会因为jar包内含的Hadoop类和集群自带的类冲突跑出一堆难以定位的ClassNotFoundException或LinkageError。这个问题在避坑章会再展开一次因为它在真实环境里出现的频率实在太高了。3.2 打包、上传与提交从本地jar到集群作业的关键命令代码写完之后打包、上传、提交三步缺一不可。打包用Maven目标是把WordCount这个类连同它依赖的第三方库打成一个可运行的jar。由于我们把Hadoop依赖设为了providedjar包只包含我们自己写的类体积很小提交前检查一下大小就能发现有没有打错依赖。# 在项目根目录下跳过测试加速打包 mvn clean package -DskipTests # 打包产物在 target/ 目录下 ls -lh target/*.jar打包完成后把jar包上传到集群的某个节点上然后用hadoop jar命令提交。这个命令会让YARN为作业分配容器启动ApplicationMaster再由它向ResourceManager申请Map和Reduce的容器资源。# 提交作业hadoop jar jar包 主类全限定名 输入路径 输出路径 hadoop jar wordcount-1.0.jar com.example.WordCount /input/words.txt /output/wc_result执行这条命令后终端会滚动输出作业的进度。关键信息有两类第一是Map和Reduce的进度百分比以及completed的task数第二是作业完成后的统计信息比如Map input records、Reduce output records。如果你的输入文件是3个单词的简单文本reduce output records就应该是3这是验证数据流的第一个信号。不能用jps查看进程状态但那不在我们现在的讨论范围。3.3 验证输出从part-r-00000里读出结果的意义作业跑完后输出目录下会出现一个或多个part-r-00000、part-r-00001这样的文件文件名里的r表示Reduce任务输出。如果Reduce数量是1就只有一个part文件如果设置了多个Reduce就有多个part文件并且顺序不一定有意义。# 查看输出目录下的文件列表 hadoop fs -ls /output/wc_result # 查看part文件内容 hadoop fs -cat /output/wc_result/part-r-00000输出内容是每行一个单词加一个制表符加一个数字比如hadoop\t42。这个格式可以直接被后续的处理工具读取。你可能想用sort -k2 -rn把结果按频率从高到低排列这是词频统计最常见的后续处理方式但在Hadoop里有一个更省事的方案把排序交给MapReduce的Shuffle阶段按key排序后再用一个只有Mapper的作业做一次反转。这个技巧我在最后一章再展开。验证输出时还有一个容易被忽略的点作业成功跑完不直接等价于结果正确。一个典型的错误是因为设置了CombinerReducer的输入记录数会下降但这个数量变化是正常的。真正需要检查的是Reduce output records是否等于预期的不重复单词数。如果输入文本里大小写不同比如Hadoop和hadoop默认的词频统计会把它们当成两个不同的词。这在“完整版”里通常会被归到数据清洗的问题是否要先转小写是否要去掉标点这些决策要在Mapper里提前处理否则后面统计出来的词频很“脏”没法直接用。4. 本地能跑、集群翻车词频统计的6个高频坑与排查方法4.1 输出目录已存在FileAlreadyExistsException为什么反复出现现象第二次运行同一个作业时提交命令立刻报错错误信息类似Output directory hdfs://.../output/wc_result already exists作业直接失败。原因这是Hadoop的主动保护机制。输出目录已存在时作业不会自动覆盖或追加它怕你误删了之前的重要结果。解决每次运行前先确认旧输出已经没用了然后用命令删掉。# 删除输出目录注意路径别写错删了没有后悔药 hadoop fs -rm -r /output/wc_result更稳妥的做法是在提交脚本里自动检查并删除输出路径。但要注意这个删除操作本身有风险如果脚本里的输出路径写错可能会删掉不该删的数据。我见过一个血泪案例某同学的脚本里把输出路径变量写成空值结果hadoop fs -rm -r把输入目录删了。所以在脚本里建议加一个保护条件确保路径包含预期的目录层级而不是根目录或空路径。4.2 jar包依赖冲突提交时报了一堆ClassNotFound现象本地用mvn package打完包上传集群后hadoop jar提交运行时抛异常异常栈里有ClassNotFoundException指向Hadoop自己的类比如org.apache.hadoop.mapreduce.Job。原因pom里的Hadoop依赖没有加providedscope导致打包时把Hadoop相关的类也打进了jar。提交时YARN会用用户jar包里的类但这些类和集群自带的类版本不一致于是引发各种RuntimeException和NoSuchMethodError。解决确认pom里所有Hadoop相关依赖的scope值都是provided。修改后重新打包检查jar包体积如果从几十MB降到几KB到几百KB就说明被打进去的Hadoop依赖排除了。另外注意如果你用了第三方工具类库比如Guava而Hadoop也依赖Guava版本冲突一样会出问题。可以用mvn dependency:tree查看依赖树看看哪些传递依赖被引入了再做排除。4.3 Reduce任务数影响的不只是性能还有文件数量现象作业跑完了输出目录下出现了几十个part文件很难一次性cat看完后续处理要逐个读文件。而有些场景下用户希望所有结果在一个文件里。原因Reduce任务的数量决定了part文件的数量。每个Reducer会独立写一个part-r-xxxxx文件。默认Reduce数是1如果代码或命令行里设置了-D mapreduce.job.reduces10输出就是10个文件。注意在Hadoop 3里这个参数名的旧写法mapred.reduce.tasks也会生效但新API推荐用mapreduce.job.reduces。解决如果你只是想看结果用hadoop fs -cat或getmerge。生产环境里如果用Hive或Spark SQL读这个目录多文件没有问题如果你需要单一文件要么把Reduce设为1数据量大时会拖慢整体时间要么在作业完成后做一次文件合并。但不要轻易删掉part文件这些文件是分布式计算的结果很多下游工具包括Hive的external table会把目录里所有part文件都当数据读删了会导致数据不完整。4.4 Windows下开发提交到Linux集群换行符与路径分隔符的坑现象在Windows上用IDEA写代码、本地调试通过把代码交给运维部署到Linux集群后MapTask在处理某些文本时结果不对或者读取输入文件时报错。原因一是代码里的路径用了Windows风格的\分隔符比如硬编码了\\input\\data.txt在Linux上解析会失败二是如果代码在Windows上生成了一些配置文件换行符是\r\n在Linux上部分工具解析会出现问题。解决代码里所有HDFS路径统一用/分隔符例如/input/data.txt不要用\。文本文件换行符问题不建议在代码里做处理更通用的做法是在数据预处理阶段统一转成Linux换行符或者直接用hadoop fs -put上传时确认源文件本身的格式。如果整个pipeline都是自己搭的建议在Mapper里加一个简单的清洗逻辑把\r去掉。private Text word new Text(); // 在map方法里切分前先移除行尾的\r String line value.toString(); if (line.endsWith(\r)) { line line.substring(0, line.length() - 1); } StringTokenizer itr new StringTokenizer(line);这段逻辑看起来很小却能把Windows和Linux换行符差异引入的脏数据挡在入口处。如果你想统计的文本里本来就包含\r作为独立字符这个逻辑按需使用别盲目照抄。4.5 小文件过多导致Map数量爆炸从根源上控制输入分片现象输入目录下放了10000个微小文件每个只有几KB作业启动后MapTask数量激增到一万多集群资源瞬间被打满整个作业跑得比预期慢很多。原因TextInputFormat的切片逻辑会为每个输入分片创建一个Map任务。默认情况下一个小文件自成一个分片不会和其他文件合并。对小文件来说Map任务的启动和调度开销远超它本身的处理耗时这是典型的“启动成本淹没计算成本”。解决在数据落地时尽量把文件合并成大块再喂给Hadoop。HDFS默认块大小是128MB单个文件如果不到块大小的1/10就应该考虑合并。但如果数据已经在HDFS上了合并的成本又很高还有一种临时方案用CombineFileInputFormat代替TextInputFormat它能把多个小文件合并到一个分片里在不移动数据的情况下降低Map数量。不过这个方案对代码侵入性较大需要自定义InputFormat类更适合进阶场景。在词频统计里我一般建议先统计一下输入目录的文件大小分布如果大量文件小于几十MB就先做一轮合并再跑统计而不是急着调整集群参数。4.6 内存设置不当导致任务被杀现象Map或Reduce阶段出现大量Container killed日志里可以看到Kill container_xxx because it exceeds memory limits。原因YARN会给每个容器分配内存上限默认的物理内存限制是虚拟内存的1.1倍左右。如果Mapper或Reducer处理的数据量超出容器内存上限比如用了大量String.split()导致对象过多或者Reducer侧拉取的数据量超过了堆内存设置就会被NodeManager强制杀掉。解决先调代码再调参数。代码层面减少不必要的对象创建比如复用Text和IntWritable实例用循环内的局部变量时注意逃逸对象。参数层面调大Map和Reduce容器的内存上限。阈值要根据你的数据量和集群情况来调不要一次性调得过大容易把NodeManager的物理内存占满反而影响同一节点上的其他作业。# 提交时临时调整容器内存注意这两个参数名在Hadoop 3.x下的写法 hadoop jar wordcount.jar com.example.WordCount /input /output \ -D mapreduce.map.memory.mb2048 \ -D mapreduce.reduce.memory.mb4096注意这种写法的适用范围它只在当前作业生效不会修改集群默认配置。生产环境里如果多个作业都需要更大的内存应该在yarn-site.xml里统一调yarn.nodemanager.resource.memory-mb和yarn.scheduler.maximum-allocation-mb。这两个配置是节点的总内存和最大容器内存改的时候要留出操作系统和Hadoop本身需要的内存。5. 参数调优4个让词频统计从“能跑”变“能打”的关键设置5.1 调整Reduce数量计算并行度与输出文件的平衡Reduce数量的设置不是拍脑袋决定的它既要考虑数据量也要考虑下游消费方。在词频统计中Reducer要做的工作就是把相同单词的计数加总Shuffle阶段已经帮我们完成了分组和排序Reducer本身的计算几乎都是轻量的真正的瓶颈往往在于拉取Map输出带来的网络传输。设置Reduce数量的大致经验公式是每个Reduce任务处理的输入数据量在256MB到1GB之间比较合适。比如输入是1GB文本Map阶段可能产生更大量的中间输出Reduce设为4到16个比较合理。但如果只是几MB的测试数据Reduce设为1反而更快因为多个Reducer之间还要做数据分区和shuffle开销大于并行收益。hadoop jar wordcount.jar com.example.WordCount /input /output \ -D mapreduce.job.reduces4这个参数需要在提交时指定。如果代码里用job.setNumReduceTasks(4)写死那么命令行参数就不会覆盖它——这一点是很多人的知识盲区。我建议把Reduce数量当作作业配置放在命令行传参而不是写死在代码里这样同一个jar包可以根据不同数据规模灵活调整不用改代码重新打包。5.2 Combiner的取舍什么时候用、什么时候不能复用Reducer词频统计里Combiner直接复用Reducer类因为求和满足交换律和结合律。但并不是所有作业的Reducer都能这样用。比如你要统计的是每个单词的平均长度Reducer逻辑是先求和再除以计数这个逻辑不可交换直接复用会导致Combiner算出平均值后又拿去给其他Combiner算平均值结果完全错误。另一种情况是设置Combiner虽然减少了shuffle数据量但增加了Map端的计算量。如果每个MapTask的输出本身不大比如几千条Combiner带来的收益很小反而因为多了一次遍历浪费CPU。我的经验是在不确定时先做一次基准测量对比开不开Combiner的作业运行时间和输出数据量。词频统计这种高重复度的场景Combiner几乎总是值得开的因为文本中同一个单词在不同行出现的频率很高Map端合并的效果立竿见影。5.3 开启Map端输出压缩当Shuffle数据量成为瓶颈时如果你处理的文本数据量到了GB级别或者集群的网络带宽比较紧张可以考虑开启Map端的输出压缩。压缩发生在Shuffle阶段写盘和传输时Reducer会透明解压对代码逻辑没有影响。代价是消耗一点点CPU时间但能显著减少磁盘IO和网络传输。hadoop jar wordcount.jar com.example.WordCount /input /output \ -D mapreduce.map.output.compresstrue \ -D mapreduce.map.output.compress.codecorg.apache.hadoop.io.compress.SnappyCodecSnappy压缩速度很快压缩比适中适合Map端中间数据。有的同学喜欢用Gzip它的压缩比更高但是CPU开销大Map端的Shuffle数据往往不需要这么高的压缩比。注意这里是map.output.compress只压缩Map和Reduce之间的数据如果你的最终输出也想压缩则需要设置mapreduce.output.fileoutputformat.compress这两个参数经常被搞混。5.4 控制内存参数与JVM重用避免老年代GC拖垮整个作业词频统计在数据量大时Mapper端最常见的性能问题是字符串对象的产生。StringTokenizer每次切分出新的String对象这些对象很快成为不可达对象触发频繁的Young GC严重的话会进入老年代GC表现为MapTask运行得越来越慢日志里的GC时间占比接近几十秒。一种做法是在代码层面尽量复用可变对象比如用Text.set()方法重置内容而不是新建Text。另一种办法是调整Map和Reduce容器的JVM参数。Hadoop的MapReduce框架允许通过提交参数控制容器的JVM内存设置但企业集群中这些参数通常由平台统一管理建议确认你的账号是否有覆盖权限否则改了不生效还会困惑很久。YARN的JVM重用设置mapreduce.job.jvm.numtasks表示一个JVM能运行多少个任务后再被销毁。如果设置为-1表示JVM可以复用同一个作业中的所有任务。这能显著减少JVM启动次数——每个JVM从启动到可运行都有秒级延迟对大量小Map任务来说JVM启动耗时的占比很高。hadoop jar wordcount.jar com.example.WordCount /input /output \ -D mapreduce.job.jvm.numtasks10但这个参数只对同一作业内的任务有效。如果你在同一个代码里连续提交多个job每个job的JVM是独立的这个参数控制不了跨job的复用。理解了它的作用范围就不会因为设置了预期外的值而产生困扰。6. 把词频统计改造成可复用的文本分析管道一个高频词提取的进阶实践到这里你已经掌握了标准的词频统计作业和它的调优手段。现在我想给你一个可以落地的进阶方向把词频统计改造成一个“高频词提取管道”让它不止能统计还能直接输出你想要的Top N高频词。改造的思路分成两步。第一步是数据清洗在Mapper里加入停用词过滤和大小写归一化。停用词是the、and、is这类在文本中大量出现但没有实际分析价值的词不把它们过滤掉统计出来的Top N永远被这些词占满。大小写归一化就是把Hadoop和hadoop统一成小写再统计这对频率排序是必要的。第二步是排序用MapReduce按频率降序输出。Shuffle阶段只按key排序不能按value排序所以要先做一个反转作业——把单词, 次数变成次数, 单词再让框架按次数排序。但直接反转有一个bug次数相同的单词会合并在一起只能看到一个单词其他的丢掉了。我在某图像处理Demo里被这个问题坑过一次后来是用自定义WritableComparable解决排序和去重问题的批处理任务里把复数词频统计要做排序这是一个绕不开的设计决策。// 自定义WritableComparable同时包含单词和次数优先按次数降序次数相同按单词升序 public class WordFreq implements WritableComparableWordFreq { private Text word new Text(); private IntWritable freq new IntWritable(); public void set(String w, int f) { word.set(w); freq.set(f); } Override public int compareTo(WordFreq o) { int cmp -1 * Integer.compare(this.freq.get(), o.freq.get()); // 降序 if (cmp ! 0) return cmp; return this.word.compareTo(o.word); // 次数相同时单词字典序 } Override public void write(DataOutput out) throws IOException { word.write(out); freq.write(out); } Override public void readFields(DataInput in) throws IOException { word.readFields(in); freq.readFields(in); } }在Reducer里把这个WordFreq设为输出keyvalue直接设成NullWritable。这样就可以用框架的排序机制对频率做全局降序排序最后的输出结果前几行就是高频词。这个自定义的比较器也顺便解决了“如何让一个数据对象可比较”的问题——后续做Top N、做推荐系统的候选集排序思路都是一样的。跑完管道后验证结果的方式很简单输出文件里的前10行应该就是你期望的高频词而不是停用词。你可以用下面的命令快速验证hadoop fs -cat /output/topn/part-* | head -n 10这个命令直接把多个part文件的前10行拼出来省去先合并再查看的步骤。如果你对键值序列化理解得比较熟还能在这个基础上加一个Partitioner来按字母区间分区或者用TotalOrderPartitioner做全局有序输出这些都是词频统计这个经典入门题通向生产级应用的延伸方向。回过头看从最基础的WordCount三行逻辑到能处理Top N的高频词管道整个链路的核心其实不在代码本身而是对MapReduce数据流的把握Map端如何预处理、Shuffle如何分发和排序、Reduce端如何聚合。我当年在这个项目上浪费最多时间的不是报错本身而是报错之后不知道从哪个日志去定位问题。所以最后一句话想跟你说的不是什么技巧而是一个习惯无论如何把每个作业的syslog和stderr日志都翻一遍再说。希望帮到你。本文还有配套的精品资源点击获取