- 正确答案：Hadoop MapReduce（MR）执行过程严格划分为5个核心阶段：Input Splitting（输入分片）、Map阶段、Shuffle and Sort（洗牌与排序）、Reduce阶段、Output Writing（输出写入）。其中，Shuffle and Sort又可细分为Partitioning（分区）、Sorting（排序）、Grouping（分组）、Combiner（可选合并）等子阶段。输入是存储在HDFS上的原始数据文件（如文本、序列化二进制等），以键值对形式抽象为对，由InputFormat（如TextInputFormat）解析生成；典型输入键为LongWritable（字节偏移量），值为Text（一行文本）。输出是Reduce阶段产生的最终结果，同样为对，由OutputFormat（如TextOutputFormat）写入HDFS，通常为纯文本格式（每行形如"key\tvalue"），也可为SequenceFile、Avro等结构化格式。 - 解答思路：面试官问“分为几个阶段”，不能仅答“Map和Reduce两个阶段”——这是常见误区。需体现对MR执行生命周期的系统性理解：从数据读取前的准备（InputSplit），到计算逻辑（Map/Reduce），再到中间数据流转（Shuffle），最后到落盘（Output）。每个阶段对应特定组件（如RecordReader、Partitioner、Sorter、Reducer等），其输入/输出类型和数据流向必须明确。同时要强调“输入/输出”在MR语境中特指Job运行时的数据抽象（即Mapper/Reducer的参数类型），而非物理文件本身。 - 深度知识讲解：（1）Input Splitting阶段： HDFS将大文件按block（默认128MB）切分，但InputSplit是逻辑切分单位，不一定与block对齐（如一个split可跨block，但不可跨文件）。TextInputFormat通过LineRecordReader逐行读取，生成作为Mapper输入。Split大小由minSize/maxSize和blockSize共同决定，可通过mapreduce.input.fileinputformat.split.minsize配置。底层依赖FileSystem.listStatus()获取文件元信息，再调用getSplits()生成Split数组。（2）Map阶段： Mapper继承自Mapper类，用户重写map()方法。输入类型K1/V1由InputFormat决定（如LongWritable/Text），输出类型K2/V2由用户指定（如Text/IntWritable）。Map任务在YARN Container中运行，输出暂存于本地磁盘（非HDFS）的环形缓冲区（默认100MB，由mapreduce.task.io.sort.mb控制）。当缓冲区达80%阈值时触发Spill（溢写），溢写前对按K2进行内存排序（Timsort），并执行Combiner（若配置且逻辑安全）以减少网络传输量。（3）Shuffle and Sort阶段（MR最复杂环节，发生在Map端和Reduce端之间）： • Map端Shuffle：多个Spill文件经归并排序（MergeSort）生成一个分区有序文件（part-r-00000），每个分区对应一个Reduce任务（由Partitioner决定，如HashPartitioner对K2.hashCode()取模）。 • Reduce端Shuffle：Reduce Task通过HTTP向各Map Task拉取属于自己分区的数据（Fetch），边拉边合并（Merge on Fetch），最终形成按K2全局排序的输入。此过程涉及： - Partitioning：Partitioner接口实现，默认HashPartitioner，要求K2实现hashCode()和equals()，否则导致数据倾斜或丢失； - Sorting：基于K2的Comparable接口（如Text.compareTo()按字典序），使用QuickSort或Timsort（小数组用InsertionSort）； - Grouping：GroupingComparator控制“哪些K2视为同一组”，默认与K2的compareTo一致，但可自定义（如按订单ID分组，忽略时间戳字段）； - Combiner：本质是本地Reduce，必须满足结合律和交换律（如sum、max），且输入输出类型与Mapper一致（K2,V2 → K2,V2），不改变逻辑语义。（4）Reduce阶段： Reducer接收>，用户重写reduce()方法。注意：V2的Iterable是迭代器而非List，底层基于堆外内存流式读取，不可重复遍历或随机访问。Reduce输出K3/V3类型由用户指定，最终交由OutputFormat处理。（5）Output Writing阶段： OutputFormat负责将Reduce输出写入HDFS。TextOutputFormat将K3.toString()+"\t"+V3.toString()写入文件；SequenceFileOutputFormat则写入二进制键值对，支持压缩（如SnappyCodec）。输出路径必须不存在（避免覆盖），由Job.setOutputPath()设置。Writer通过RecordWriter（如LineRecordWriter）完成实际写入，底层调用FSDataOutputStream.write()。补充底层机制： • 序列化：MR全程使用Writable接口（而非Java Serializable），因后者反射开销大且不支持跨语言。所有类型（Text、IntWritable等）需实现write(DataOutput)和readFields(DataInput)，支持零拷贝（如Text.getBytes()直接返回byte[]引用）。 • 推测执行（Speculative Execution）：对慢Task启动备份副本，谁先完成谁生效，通过mapreduce.map.speculative控制。 • 容错：Map失败则重试（max 4次），Reduce失败则重新拉取全部Map输出；TaskTracker/NodeManager崩溃时，ApplicationMaster重启失败Task。 - 伪代码（简化版MR执行流程）： ``` // 1. Input Splitting (by InputFormat) List splits = inputFormat.getSplits(job); for each split in splits: RecordReader reader = inputFormat.createRecordReader(split, context); while (reader.nextKeyValue()): mapContext.write(k2, v2); // mapper output // 2. Map phase (in memory buffer) RingBuffer buffer = new RingBuffer(100MB); if (buffer.isFull(80%)): spillToDisk(sortByKey(buffer), combinerIfSet()); // 3. Shuffle & Sort (distributed) // Map side: merge spills → part-r-00000 (partitioned & sorted) // Reduce side: fetch partitions → merge on disk → sort globally → group by key // 4. Reduce phase for each > in groupedInput: reduceContext.write(k3, v3); // reducer output // 5. Output Writing OutputFormat outputFormat = job.getOutputFormat(); RecordWriter writer = outputFormat.getRecordWriter(outputPath); writer.write(k3, v3); writer.close(); ```

京东数据开发实习一面

全站热榜

创作者周榜