★ FEATURED ARTICLE MapReduce做词频统计并把出现次数最多的前十个单词挑出来这个实验我前后做了很多遍也帮人排查过不少作业里的报错。今天把这套流程从头拆到尾说清楚为什么不能再用单机思维写代码Mapper和Reducer到底该怎么设计以及跑起来之后那些文档里不会写的坑。这篇不是复述官方文档而是把我在真实环境里踩过的坑、改过的代码、验证过的方法按顺序写下来。实验目标很明确对一个文本数据集做词频统计然后只保留前十个高频词输出成文件。整套流程跑通之后你能看到MapReduce从输入、拆分、洗牌到归约的完整闭环也能自己在结果上做核对。1. 为什么“统计词频挑Top10”看起来简单却值得用MapReduce重做一遍1.1 数据规模的质变才是这个问题的根源先看单机脚本大概是什么样子。在数据量不大的时候词频统计确实是最没技术含量的事情from collections import Counter counter Counter() with open(input.txt, encodingutf-8) as f: for line in f: for word in line.split(): counter[word] 1 for word, count in counter.most_common(10): print(word, count)这段代码遍历一次文件用哈希表累加内存里最多就是“不同单词数 × 键值对占用”。几万字、几百万字、甚至几百MB文本都能扛住。可一旦数据量上升到几十GB、几百GB并且存放在几十台机器的分布式文件系统上单机脚本就同时撞上两堵墙内存墙和IO墙。哈希表要装下所有词的计数内存不够要把所有数据汇聚到一台机器上网络传输和磁盘读取时间又不可接受。这时候MapReduce的核心价值才真正体现出来。它不是更聪明的算法而是把“分而治之”做成了标准流水线。数据分散在哪儿计算就推到哪儿进行第一轮加工每个节点只需要处理自己本地的那一小块数据输出中间结果再由框架完成分组与合并。理论上同样的词频统计逻辑可以线性扩展到更大的数据量。1.2 词频统计为什么是理解MapReduce的最佳样例一个MapReduce作业只有三个主要阶段map、shuffle、reduce。拿词频统计来对照每个阶段都能踩在模型上map阶段读取输入分片把一行文本拆成单词输出单词, 1。shuffle阶段框架按单词做排序、分组、分发同一个单词的所有计数会被集中到同一个reduce任务。reduce阶段对每个单词把收到的整数序列累加得到总次数。所以我的观点一直是别把这个实验当成“装个Hadoop跑个wordcount”那么简单。它会强迫你理解Mapper的输出key如何决定分组Combiner在哪里提升性能Partitioner怎么影响数据倾斜以及最终输出文件为什么会有多个part。这四件事弄明白了之后你写任何MapReduce作业都会顺手很多。1.3 用伪分布式还是真实集群初次做实验我建议先搭伪分布式也就是单机同时扮演HDFS、YARN和MapReduce执行的角色。它的优点是部署简单、调试直观能直接打开浏览器看任务进度缺点是shuffle阶段的数据只是走本地磁盘和回环网络看不到真正的跨节点传输效果。如果你手头有现成的真实集群做对比实验会更明显但课程作业和大部分入门练习用伪分布式完全足够。2. 环境和测试数据准备实验效果的差距往往从准备阶段就拉开了2.1 Hadoop环境里这几个配置会影响词频统计搭建环境时几个关键配置直接影响后续作业能不能顺利跑完。Core、HDFS、YARN三个配置文件各有一个项值得重点关注。配置文件关键项建议值说明core-site.xmlfs.defaultFShdfs://localhost:9000指定NameNode地址hdfs-site.xmldfs.replication1伪分布式单机存多副本意义不大且浪费空间yarn-site.xmlyarn.nodemanager.resource.memory-mb根据机器内存调整影响任务并行度和内存分配上限这里有一个新手很容易踩的坑NameNode的格式化只能在首次执行时做。一旦格式化原来的数据目录就清空了。所以如果你只是在做实验没有什么历史数据要保留这个问题不明显但如果你已经往HDFS里上传过文件千万别手滑再格式化一次。启动命令一般是这样# 启动HDFS和YARN start-dfs.sh start-yarn.sh # 查看进程是否正常 jps正常会看到NameNode、DataNode、SecondaryNameNode、ResourceManager、NodeManager这几个进程。少一个都要先排查日志不要急着跑作业。Hadoop的报错信息有时候很绕但大部分启动问题都能在logs/目录里找到直接原因。2.2 测试数据要故意做出“头部效应”很多初学者随手从网上复制一段文章丢进去结果Top10里全是the、of、and这类词跑出来的结果既没有区分度也很难判断代码是否正确。更推荐的做法是构造一份带有明显分布特征的测试数据。我在实验中会用Python脚本生成两千万行文本让一部分词占绝对大头比如状态码“200”出现频率远高于其他词一部分词保持中等频次再留一批低频长尾词。这样一个Top10结果能明显看出谁该在前面谁该被淘汰。import random # 模拟日志行关键词权重有明显分层 freq_words [ (200, 500000), (404, 200000), (500, 80000), (GET, 300000), (POST, 120000), ] long_tail [debug, timeout, retry, cache, session] * 20 with open(sample.log, w, encodingutf-8) as f: words [] for w, cnt in freq_words: words.extend([w] * cnt) words.extend(long_tail * 5000) random.shuffle(words) for i in range(0, len(words), 10): f.write( .join(words[i:i10]) \n)这里的关键不是模拟得要多真实而是让结果可预测。你事先就知道“200”大约出现多少次和谁在同一量级Top10应该包含哪几个词。这样实验结束才能立刻发现reduce逻辑里是否发生了丢数据或者统计错误。如果随机种子不固定每次生成的数据分布都不一样后续对照结果会非常痛苦。生成好文件后把它放到HDFS上hdfs dfs -mkdir -p /user/input hdfs dfs -put sample.log /user/input/ hdfs dfs -ls /user/input3. Mapper与Reducer的代码实现每一行都有取舍3.1 Mapper在分词时需要考虑的边界MapReduce的Mapper输入是LongWritable, Textkey是行偏移量value是行内容。这里容易踩的第一个坑是空行和非法字符。文件末尾常常有一个空行如果不对空行做过滤会把空字符串作为key输出虽然不影响Top10排序却会让Counter统计和后续调试多出不少干扰。分词规则我用正则[^a-z0-9]来清洗英文单词把逗号、句号、引号全部过滤掉。如果实验数据本身是中文最简单的方式是先用空格或标点把句子切分成词再把切分结果作为输入MapReduce本身不做中文分词这块需要前置预处理。下面是我实验里用到的Mapper实现public class WordCountMapper extends MapperLongWritable, Text, Text, LongWritable { private static final LongWritable ONE new LongWritable(1L); private Text outKey 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[] words line.toLowerCase().split([^a-z0-9]); for (String w : words) { if (w.isEmpty()) { continue; } outKey.set(w); context.write(outKey, ONE); } } }这里为什么要toLowerCase()因为统计词频的实验里The和the通常应当被看作同一个词。这种归一化看似简单却决定了统计口径不统一大小写Top10里会出现重复词根没法跟预期结果对上。还有一个性能细节new Text()和new LongWritable()最好不要在map循环里反复创建尤其是数据量大时每行几十个单词都会触发对象创建GC压力会被放大。把输出key复用成一个成员变量只调用set更新内容是Job性能优化里成本最低的一招。3.2 Reducer端求TopN全排序还是局部TopNReducer接收到的数据是key某个单词values一组1的迭代器。最直白的做法是先把所有单词的频率都reduce输出再用另一个排序程序取前十个。但这样做意味着shuffle阶段要把每一个单词的计数都传出来当单词量很多时中间结果会被放大好几倍。更聪明的做法是在每个Reducer内部只维护一个容量为10的TreeMap只保留当前出现次数最高的10个词。举个例子假设某个Reducer分到了100万个单词的记录它根本不需要记住那999990个低频词的完整计数只需要在迭代过程中不断和当前的最弱候选比较即可。但这里有一个很容易中招的细节如果TreeMap的key直接用次数遇到频次相同的两个词后插入的词会被当作重复key丢掉。我第一版代码就是这个写法结果跑出来发现明明该有20个词的候选池最后只剩了10个不到。正确的做法是让比较器同时考虑次数和单词本身public static class TopNReducer extends ReducerText, LongWritable, Text, LongWritable { private TreeMapString, Long topN new TreeMap(new ComparatorString() { Override public int compare(String a, String b) { // 先按次数降序其次按单词字典序避免TreeMap吞词 int cmp Long.compare(topN.getOrDefault(b, 0L), topN.getOrDefault(a, 0L)); if (cmp 0) { return a.compareTo(b); } return cmp; } }); Override protected void reduce(Text key, IterableLongWritable values, Context context) throws IOException, InterruptedException { long sum 0L; for (LongWritable v : values) { sum v.get(); } topN.put(key.toString(), sum); if (topN.size() 10) { String lastKey topN.lastEntry().getKey(); topN.remove(lastKey); } } Override protected void cleanup(Context context) throws IOException, InterruptedException { for (Map.EntryString, Long e : topN.entrySet()) { context.write(new Text(e.getKey()), new LongWritable(e.getValue())); } } }比较规则是“次数高的往前排次数相同时再看单词本身”这样TreeMap中最后一个元素就是当前Top10里最弱的那一个超过容量时直接移除它。通过这种局部TopN的写法中间结果不会膨胀每个Reducer只需要维护10个词条内存占用基本是常量级别。3.3 Driver设置Job配置里容易被忽略的三处主类里的Driver负责装配整个作业。我常看到配置不完整导致的结果“莫名少词”多半是下面这三个地方出问题。第一输出目录不能预先存在。如果上一次运行留下的output目录还在Hadoop会直接报FileAlreadyExistsException。所以在提交前要么手动删除要么在Driver里先判断并删除Path outputPath new Path(args[1]); FileSystem fs FileSystem.get(conf); if (fs.exists(outputPath)) { fs.delete(outputPath, true); }第二作业名称和Jar类要成对设置。job.setJarByClass()和job.setJobName()别漏掉不然在YARN的UI里看到一堆默认名字排错时很难区分是哪一次提交的任务。第三Reducer数量不要随意设成1。对TopN实验来说如果Reducer数量为1等于所有单词都汇聚到一个节点逻辑上虽然简单却违背了并行统计的初衷也失去了观察数据倾斜的机会。更合理的做法是设置多个Reducer并行每个Reducer局部Top10再在Job之外写一段合并逻辑。Driver的参考代码如下public class WordCountTopNJob { public static void main(String[] args) throws Exception { Configuration conf new Configuration(); Job job Job.getInstance(conf, word count top n); job.setJarByClass(WordCountTopNJob.class); job.setMapperClass(WordCountMapper.class); job.setCombinerClass(TopNReducer.class); job.setReducerClass(TopNReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(LongWritable.class); FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }这里我把Combiner直接复用TopNReducer是因为combine本质上就是对当前map输出的一份小范围“局部聚合”它和reduce处理的数据结构相同。对于词频统计来说这种复用完全可行能大幅减少shuffle传输量。还需要想清楚一点当Reducer数量大于1时输出目录里会有多个part文件每个文件的Top10只是局部的。要得出全局前10最简单的方法是把part文件合并后用一行脚本排序也可以再提交一个只有Map任务、不设Reducer的小作业让Map读入所有局部Top10并全局排序。这个逻辑别放在Reducer里硬做否则会误导你故意把Reducer数量设成1反而丢失了并行度。4. 打包、提交与参数调整让作业真正跑起来4.1 用Maven构建jar包工程结构最好一开始就规整好pom.xml src/main/java/com/example/wordcount/ WordCountMapper.java TopNReducer.java WordCountTopNJob.javapom.xml里引入依赖时版本最好和集群上的Hadoop保持一致。初学者最容易犯的错是本地用Hadoop 2.x的依赖编译打到3.x的集群上跑运行时各种NoSuchMethodError。我的做法是直接查集群里hadoop version按那个版本号来配Maven依赖。dependencies dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.3.6/version /dependency /dependencies构建命令只有一行mvn clean package -DskipTests。把生成的jar包上传到集群节点后提交作业前顺手确认三件事输入数据已经放到HDFS上输出目录不存在jar包在节点的本地文件系统中而不是在某个人都不了的临时目录里。4.2 常用提交命令与参数说明hadoop jar wordcount-topn.jar com.example.wordcount.WordCountTopNJob \ /user/input/sample.log \ /user/output/wc_top10跑起来之后我建议立刻去YARN的ResourceManager页面看三个指标正在运行的Mapper数、Reducer数、每个任务的处理时间。如果Mapper数量远小于输入文件块数说明输入路径指向的文件太小或者文件数太少如果Reducer有任务迟迟不结束大概率是数据倾斜。这里整理一份常用参数供参考参数设置方式影响输入分片大小FileInputFormat.setMinInputSplitSize影响Mapper数量Combinerjob.setCombinerClass压缩shuffle数据量Reduce并行度job.setNumReduceTasks影响最终part文件数量JVM堆mapreduce.map.memory.mb大Mapper时避免OOM压缩传输mapreduce.output.fileoutputformat.compress减小落盘数据需要提醒的是伪分布式环境下不要盲目调大并行度。单机总共就那么多内存和CPU设8个Reducer并不比设2个快反而会增加上下文切换和本地文件读写开销。这个实验的首要目标是正确性和可理解性能优化是下一步的事。5. 结果验证与那些“看起来对了其实错了”的坑5.1 多part文件与全局Top10的核对作业跑完后输出目录下通常有part-r-00000等多个文件。直接cat其中某一个文件看到的只是局部Top10。我在第一版实验里就吃过这个亏自信满满地打开第一个part发现“前十个单词”和自己预先计算的预期Top10对不上查了很久代码最后才发现问题是part0000只是所有Reducer里的一个。正确的核对流程先把所有part文件合并到一起再写一段聚合脚本。如果启动了3个Reducer合并后的记录数最多是30条每个Reducer留下10条其中同一个词可能出现在多个part里还需要做一轮聚合。这句话请大家记住MapReduce帮你把词频统计好了但“全局Top10”这一步实验本身并没有替你完成这是这个题目最容易让人产生误解的地方。hdfs dfs -cat /user/output/wc_top10/part-r-* | \ awk {cnt[$1] $2} END {for (w in cnt) print cnt[w], w} | \ sort -rn | head -10如果这里得到的Top10和预想不一致不要急着怀疑MapReduce先把数据生成的随机种子固定下来再用单机脚本跑一遍小样本数据的结果做对照。稳定的实验数据加稳定的校验脚本是排查一切不确定因素的前提。5.2 数据倾斜热门单词把单个Reducer打爆真实日志里词频分布往往是极度不均衡的一个“200”状态码可能出现几十万甚至上千万次。问题就在shuffle阶段会按照key的哈希把同一个单词的所有计数送给同一个Reducer这个Reducer收到的数据量可能比其他Reducer高出几个数量级。其他任务早就跑完了它还在慢慢累加那串长长的values。处理倾斜的思路有几条在map端用Combiner做一轮预聚合把同一个单词在本地节点的计数先合并成一次而不是把上亿个“1”原样传到reduce端。如果分布极端可以考虑加盐salt拆分key给热门单词加上随机后缀分布到不同Reducer先做局部统计再在二次作业中还原出真实总次数。在作业参数层面适当增加内存和重试容忍度但这只是缓解不是根治。对于这个实验的规模防倾斜的第一道防线永远是Combiner。很多教程会把Combiner介绍成“锦上添花的优化”但在我看来词频统计这种“每个key产生大量小value”的场景没有Combiner的作业在真实集群上几乎必然遇到Reducer长尾问题。5.3 中文乱码、空输出、任务失败排查速查下面这些坑是我在实践中反复遇到的整理成速查表遇到问题先对号入座症状常见原因解决思路输出中文乱码输入文件编码和Hadoop默认不一致统一UTF-8文本文件写入时不要带BOM头输出目录为空或文件大小异常输入路径写错或正则分词过滤掉了全部内容先看Counter里map input records是否为0作业提交后秒失败输出路径已存在删除输出目录或换一个新目录某些词没进Top10但预期有TreeMap比较器没处理key重复用“次数单词”共同决定排序map阶段一直重启单个Mapper处理的文件过大或内存不足调整mapreduce.map.memory.mb检查节点内存还有一个实用小技巧在Mapper和Reducer里通过context.getCounter()自定义统计项把“读取行数”“过滤空行数”“输出单词数”打印到Counter里。这些Counter会直接显示在YARN页面上肉眼可查远比你反复看日志要方便。6. 这份实验做完之后真正值得沉淀的东西6.1 wordcount里的三类思想迁移词频统计的代码量不大但它背后有三个可以迁移到任何大数据任务里的思想。第一是“分而治之”数据拆分后各自计算再合并结果第二是“本地聚合”Combiner其实就是在计算过程中尽早减少数据量这和Spark的reduceByKey是一脉相承的思路第三是“TopN不等于全排序”很多场景我们只需要最好的那一小部分没必要对所有数据做全局排序用小顶堆或候选集维护TopK是流式计算里特别常见的优化手段。6.2 换成Spark之后思路有什么变化如果你做完MapReduce再看Spark会发现很多概念都能对应上。Spark的textFile对应输入flatMap对应map端分词reduceByKey就是把Combiner和Reducer合并成一个算子。区别在于Spark可以让尽量多步骤在内存里完成避免MapReduce反复落盘在小到中等规模的数据上性能优势明显。TopN在Spark里可以先reduceByKey后takeOrdered(10)底层也用到了类似局部候选集合的机制。理解了MapReduce的这套拆解过程再学Spark、Flink都会非常快。6.3 我自己的习惯每次实验都留一次“破坏性测试”最后说一个个人习惯。我每次跑这类实验不会只拿一份规规矩矩的数据跑完就算而是至少额外准备三份“不太好”的输入一份夹了大量空行的文件一份包含非UTF-8编码字符的文件一份词频分布极度不均匀的文件。用它们分别跑一遍观察计数的Counter变化、Reducer耗时曲线和最终的Top10结果。几次下来你对框架的理解会比单纯做一遍标准实验深得多。这种“破坏性测试”的价值在于它逼着你把代码里的隐式假设暴露出来。空行处理得当吗非法字符会不会让正则漏掉词数据倾斜时Reducer会不会卡死这些问题在教科书里都有答案但只有你在自己的作业里亲手触发过一次才会真的记得住。拿它作为这个实验的收尾我觉得比任何总结都有用。
阅读完成 · 觉得有帮助？
