1. MapReduce编程模型入门第一次接触MapReduce时我被它分而治之的思想惊艳到了。想象你面前堆着100本厚厚的书籍需要统计每个单词出现的次数。单枪匹马干这活可能要几个月但如果你有100个助手每人负责一本书最后把结果汇总起来效率就完全不同了。这就是MapReduce的核心思想。MapReduce模型包含两个关键阶段Map阶段把大任务拆分成小任务并行处理Reduce阶段把各个小任务的结果合并举个生活中的例子就像餐厅后厨准备宴会厨师长JobTracker把菜单拆成冷盘、热菜、甜点等任务各位厨师Mapper并行处理自己的任务服务员Reducer把各道菜按桌号归类上菜在技术实现上MapReduce有三大神奇之处自动并行化框架自动分配计算资源容错机制某个节点挂了会自动重新调度数据本地化尽量在数据所在的节点进行计算2. 经典WordCount实战WordCount是MapReduce界的Hello World我们来解剖这只麻雀。2.1 数据准备先创建测试文件# 生成包含10万行hello hadoop的文本 for i in {1..100000}; do echo hello hadoop test.txt done # 上传到HDFS hdfs dfs -put test.txt /input2.2 Java代码实现完整代码包含三个部分// Mapper实现 public class WordCountMapper extends MapperLongWritable, Text, Text, IntWritable { private final static IntWritable one new IntWritable(1); private Text word new Text(); public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 将每行文本按空格切分 String[] words value.toString().split( ); for (String w : words) { word.set(w); // 设置当前单词 context.write(word, one); // 输出单词,1 } } } // Reducer实现 public class WordCountReducer extends ReducerText, IntWritable, Text, IntWritable { private IntWritable result new IntWritable(); 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); // 输出单词,总次数 } } // 主类配置 public class WordCount { 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(WordCountMapper.class); job.setReducerClass(WordCountReducer.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); } }2.3 运行与结果打包执行hadoop jar wordcount.jar WordCount /input /output查看结果hdfs dfs -cat /output/part-r-00000 # 输出示例 # hadoop 100000 # hello 1000002.4 内部机制图解数据流转过程原始文本 → Split分片 → Map处理 → Shuffle排序 → Reduce汇总 → 最终结果 (HDFS) (本地计算) (网络传输) (本地计算) (HDFS)3. 数据清洗实战文件合并去重实际工作中经常遇到多个数据源需要合并去重的情况。比如有两个用户行为日志文件A和B需要合并并去除重复记录。3.1 样例数据文件A内容user1,click,page1 user2,view,page2 user3,click,page1文件B内容user1,view,page3 user2,click,page2 user4,view,page13.2 核心代码实现// Mapper直接输出整行作为key public class MergeMapper extends MapperLongWritable, Text, Text, NullWritable { public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { context.write(value, NullWritable.get()); } } // Reducer直接输出key实现去重 public class MergeReducer extends ReducerText, NullWritable, Text, NullWritable { public void reduce(Text key, IterableNullWritable values, Context context) throws IOException, InterruptedException { context.write(key, NullWritable.get()); } }3.3 执行效果合并去重后结果user1,click,page1 user1,view,page3 user2,click,page2 user2,view,page2 user3,click,page1 user4,view,page14. 数据排序实战对海量数据排序是经典需求比如需要按用户访问次数降序排列。4.1 排序原理MapReduce的排序魔法发生在Shuffle阶段Map端输出会按key排序Reduce端接收的数据已经是按key有序的4.2 实现全局排序如果要实现全局有序而非仅分区内有序需要使用TotalOrderPartitioner提前采样确定key分布// Mapper提取排序字段 public class SortMapper extends MapperLongWritable, Text, IntWritable, Text { public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields value.toString().split(,); int score Integer.parseInt(fields[1]); context.write(new IntWritable(score), value); } } // Reducer直接输出 public class SortReducer extends ReducerIntWritable, Text, Text, NullWritable { public void reduce(IntWritable key, IterableText values, Context context) throws IOException, InterruptedException { for (Text val : values) { context.write(val, NullWritable.get()); } } }4.3 优化技巧二次排序先按字段A排序字段A相同再按字段B排序内存优化对于大数据量使用外部排序避免OOM5. 性能调优实战在真实集群运行MapReduce作业时我踩过不少性能坑总结几个关键点5.1 资源配置!-- mapred-site.xml配置示例 -- property namemapreduce.map.memory.mb/name value2048/value /property property namemapreduce.reduce.memory.mb/name value4096/value /property5.2 Combiner使用在WordCount例子中可以在Map端先本地聚合job.setCombinerClass(WordCountReducer.class);这样能减少Shuffle数据传输量。5.3 数据倾斜处理对于热key导致的数据倾斜可以采用增加Reducer数量自定义Partitioner分散热key两阶段聚合先给key加随机前缀局部聚合再去前缀全局聚合6. 实际应用场景MapReduce虽然现在有更快的替代方案但在某些场景依然不可替代6.1 离线日志分析用户行为分析广告点击统计异常检测6.2 数据预处理ETL流程特征工程数据清洗6.3 复杂计算PageRank算法推荐系统协同过滤图计算7. 常见问题排查根据我处理过的线上问题总结几个典型case7.1 作业卡住可能原因资源不足检查YARN资源队列数据倾斜查看Counter中的RECORDS分布网络问题检查节点间连通性7.2 性能瓶颈定位方法# 查看作业执行详情 mapred job -history all job_id7.3 内存溢出解决方案增大container内存优化数据结构减少缓存数据量8. 进阶技巧8.1 自定义InputFormat处理特殊格式数据时可以继承FileInputFormatpublic class JsonInputFormat extends FileInputFormatLongWritable, MapWritable { // 实现RecordReader解析JSON }8.2 分布式缓存共享小文件job.addCacheFile(new URI(/data/cities.txt#cities));8.3 计数器使用统计异常记录context.getCounter(Error, BadRecord).increment(1);9. 与Spark对比虽然Spark现在更流行但理解差异很重要特性MapReduceSpark计算模型批处理批/流/交互式执行速度慢(需落盘)快(内存计算)易用性较复杂高阶API适用场景超大数据量迭代计算10. 最佳实践建议小文件合并使用CombineTextInputFormat处理大量小文件压缩中间数据配置mapreduce.map.output.compress合理设置Reducer数建议为节点数的0.95~1.75倍重用JVM设置mapreduce.job.jvm.numtasks记得第一次成功运行MapReduce作业时看着控制台刷新的日志那种成就感至今难忘。虽然现在处理同样需求可能会选择Spark但掌握MapReduce的底层原理依然是每个大数据工程师的必修课。当你真正理解Mapper和Reducer如何协同工作面对更复杂的分布式系统时也会更加得心应手。