Apache Hadoop分布式计算模型：MapReduce任务调度与执行流程解析

平荔允Imogene

461人浏览 · 2025-09-08 19:06:29

平荔允Imogene · 2025-09-08 19:06:29 发布

Apache Hadoop分布式计算模型：MapReduce任务调度与执行流程解析

【免费下载链接】hadoop Apache Hadoop 项目地址: https://gitcode.com/gh_mirrors/ha/hadoop

引言：从数据洪流到计算集群——MapReduce的核心价值

在大数据时代，企业面临着PB级甚至EB级数据的处理挑战。传统单机计算模型因硬件资源限制，无法高效处理此类规模的数据。Apache Hadoop的MapReduce分布式计算模型通过将复杂任务分解为可并行执行的子任务，充分利用集群中多台计算机的计算能力，实现了海量数据的高效处理。本文将深入解析MapReduce的任务调度机制与执行流程，帮助读者理解其内部工作原理，从而更好地应用和优化MapReduce作业。

读完本文，您将能够：

理解MapReduce的核心设计思想与架构组成
掌握MapReduce作业从提交到完成的完整生命周期
深入了解任务调度策略及其对作业性能的影响
熟悉MapReduce任务执行的详细流程，包括Map阶段、Shuffle阶段和Reduce阶段
学会使用MapReduce的关键API，并通过示例代码实现自定义MapReduce作业

MapReduce核心架构与组件

MapReduce采用"主从架构"（Master-Slave Architecture），由一个主节点（JobTracker）和多个从节点（TaskTracker）组成。在YARN（Yet Another Resource Negotiator）架构中，JobTracker的功能被拆分为ResourceManager和ApplicationMaster，以提高系统的可扩展性和容错性。

MapReduce 1.0架构

mermaid

MapReduce 2.0（YARN）架构

mermaid

核心组件功能

ResourceManager：全局资源管理器，负责整个集群的资源分配与调度。
NodeManager：运行在每个节点上的代理，负责管理本节点的资源和容器生命周期。
ApplicationMaster：每个MapReduce作业对应的应用管理器，负责作业的调度和监控。
Container：资源分配的基本单位，包含CPU、内存等资源，用于运行Map或Reduce任务。
MapTask：负责处理输入数据的分片，执行Map函数，生成中间结果。
ReduceTask：负责聚合MapTask生成的中间结果，执行Reduce函数，生成最终结果。

MapReduce作业生命周期

MapReduce作业从提交到完成，经历多个阶段，涉及多个组件的协同工作。

作业生命周期概览

mermaid

作业提交与初始化

作业提交：客户端调用Job.submit()方法提交作业。客户端会先检查作业配置，如输入输出路径、Map和Reduce类等，然后将作业资源（JAR包、配置文件等）上传到HDFS，并向ResourceManager提交作业。
作业初始化：ResourceManager收到作业提交请求后，会为作业分配一个Container来启动ApplicationMaster。ApplicationMaster启动后，会加载作业配置，获取输入数据的分片信息（InputSplit），并计算所需的Map任务数量和Reduce任务数量。
任务调度与资源分配：ApplicationMaster根据数据本地化原则，向ResourceManager申请运行Map和Reduce任务所需的资源。ResourceManager根据集群资源状况，为任务分配Container。

任务执行流程

Map任务执行：
- 读取输入数据分片（InputSplit）
- 对输入数据进行解析，生成键值对（Key-Value Pair）
- 执行用户定义的Map函数，处理键值对，生成中间键值对
- 对中间键值对进行本地排序和合并（Combiner）
Shuffle阶段：
- Map任务将中间结果分区并写入本地磁盘
- Reduce任务通过HTTP协议从Map任务节点拉取属于自己的中间结果
- Reduce任务对拉取的中间结果进行合并和排序
Reduce任务执行：
- 对排序后的中间键值对执行用户定义的Reduce函数
- 将Reduce函数的输出结果写入HDFS

作业完成与清理

ApplicationMaster监控所有Map和Reduce任务的执行状态，当所有任务完成后，作业进入"成功"状态。
ApplicationMaster向ResourceManager注销自己，并释放所有资源。
客户端可以通过Job.waitForCompletion()方法获取作业执行结果。

任务调度策略

MapReduce的任务调度策略直接影响作业的执行效率和资源利用率。Hadoop提供了多种调度策略，以适应不同的应用场景。

调度策略分类

FIFO调度器（First-In-First-Out Scheduler）：
- 按照作业提交的顺序进行调度，先提交的作业优先获取资源。
- 实现简单，但可能导致长作业阻塞短作业，资源利用率不高。
容量调度器（Capacity Scheduler）：
- 将集群资源划分为多个队列，每个队列有固定的资源容量。
- 作业提交到指定队列，队列内采用FIFO策略调度。
- 支持队列资源共享，当一个队列资源空闲时，其他队列可以使用其空闲资源。
公平调度器（Fair Scheduler）：
- 动态调整资源分配，使得每个作业最终获得公平的资源份额。
- 支持作业优先级和资源抢占，确保高优先级作业能获得足够资源。

数据本地化调度

为了减少数据传输开销，MapReduce采用数据本地化调度策略：

mermaid

数据本地化级别：

节点本地化（Node-local）：Map任务与数据块在同一节点，性能最佳。
机架本地化（Rack-local）：Map任务与数据块在同一机架的不同节点，有一定网络开销。
跨机架（Off-rack）：Map任务与数据块在不同机架，网络开销较大。

ApplicationMaster在调度Map任务时，会优先选择节点本地化，其次是机架本地化，最后才是跨机架调度。

MapReduce任务执行详细流程

Map阶段执行流程

Map阶段是MapReduce处理数据的第一个阶段，负责将输入数据转换为中间键值对。

mermaid

InputSplit：输入数据被分割为多个InputSplit，每个InputSplit由一个Map任务处理。InputSplit的大小通常与HDFS块大小一致（默认128MB），以提高数据本地化。
RecordReader：负责从InputSplit中读取数据，并将其解析为键值对（Key-Value Pair）。Hadoop默认提供了多种RecordReader实现，如TextInputFormat用于读取文本文件，将每一行解析为一个键值对，键为行偏移量，值为行内容。
Map函数：用户自定义的Map函数，对RecordReader生成的键值对进行处理，生成中间键值对。例如，在WordCount示例中，Map函数将每一行文本分割为单词，并输出<单词, 1>的键值对。
Partitioner：对Map函数输出的中间键值对进行分区。默认使用HashPartitioner，通过对Key进行哈希计算来确定分区号。分区数量等于Reduce任务数量，确保相同Key的键值对被分配到同一个Reduce任务。
Sort：每个分区内的中间键值对按照Key进行排序。
Combiner：可选的本地聚合步骤，对排序后的键值对进行合并，减少后续Shuffle阶段的数据传输量。Combiner的实现通常与Reduce函数相同，但Combiner的输出结果必须与Reduce函数的输入兼容。
Spill：当内存中的中间结果达到一定阈值时，会将数据溢写到本地磁盘。
Merge：Map任务完成前，会将所有溢写文件合并为一个有序的中间结果文件，等待Reduce任务拉取。

Shuffle与Sort阶段

Shuffle阶段是MapReduce的核心，负责将Map任务的中间结果传输到Reduce任务，并进行排序和合并。

mermaid

Fetch：Reduce任务启动多个拉取线程（Fetcher），通过HTTP协议从Map任务节点拉取属于自己分区的中间结果。
InMemoryBuffer：拉取的中间结果先存储在内存缓冲区中。
SpillToDisk：当内存缓冲区达到一定阈值时，将数据溢写到磁盘。
Merge：对磁盘上的多个溢写文件进行多路归并，合并为一个有序的文件。
Sort：对合并后的文件进行排序，确保相同Key的键值对连续排列。

Reduce阶段执行流程

Reduce阶段负责对Shuffle阶段传输过来的中间结果进行处理，生成最终输出。

mermaid

Grouping：将排序后的中间键值对按照Key进行分组，将相同Key的所有Value组成一个集合。
Reduce函数：用户自定义的Reduce函数，对Grouping生成的键值对集合进行处理，生成最终的键值对。例如，在WordCount示例中，Reduce函数对相同单词的计数进行求和，输出<单词, 总次数>的键值对。
RecordWriter：负责将Reduce函数输出的最终键值对写入HDFS。Hadoop默认提供了多种RecordWriter实现，如TextOutputFormat将键值对以制表符分隔的形式写入文本文件。

MapReduce核心API与示例代码

MapReduce核心类与接口

Job：表示一个MapReduce作业，用于配置作业参数、设置Map和Reduce类、输入输出格式等。
Mapper：Map任务的基类，用户通过继承该类并重写map()方法实现自定义Map逻辑。

public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
    protected void map(KEYIN key, VALUEIN value, Context context) 
            throws IOException, InterruptedException {
        // 用户自定义Map逻辑
    }
}

Reducer：Reduce任务的基类，用户通过继承该类并重写reduce()方法实现自定义Reduce逻辑。

public class Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
    protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context context)
            throws IOException, InterruptedException {
        // 用户自定义Reduce逻辑
    }
}

Context：Map和Reduce函数的上下文对象，用于输出键值对、获取作业配置、报告进度等。
InputFormat：定义输入数据的格式，负责将输入数据分割为InputSplit，并提供RecordReader用于读取数据。常用实现有TextInputFormat、KeyValueTextInputFormat等。
OutputFormat：定义输出数据的格式，负责将Reduce函数输出的键值对写入HDFS。常用实现有TextOutputFormat、SequenceFileOutputFormat等。

WordCount示例代码

WordCount是MapReduce的经典示例，用于统计文本文件中单词的出现次数。

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 {

    // Map类：将文本行分割为单词，输出<单词, 1>
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(word, one);
            }
        }
    }

    // Reduce类：对相同单词的计数进行求和，输出<单词, 总次数>
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }

    // 主函数：配置并提交MapReduce作业
    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); // 使用Reduce类作为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);
    }
}

代码解析

TokenizerMapper：继承自Mapper类，重写map()方法。map()方法将输入的文本行（Text value）分割为单词，对于每个单词，输出<单词, 1>的键值对。
IntSumReducer：继承自Reducer类，重写reduce()方法。reduce()方法对相同单词的所有计数（Iterable<IntWritable> values）进行求和，输出<单词, 总次数>的键值对。
main方法：配置MapReduce作业。设置作业名称、Map和Reduce类、输出键值对类型、输入输出路径等。通过job.setCombinerClass(IntSumReducer.class)设置Combiner，使用Reduce类作为Combiner进行本地聚合。最后调用job.waitForCompletion(true)提交作业，并等待作业完成。

MapReduce性能优化策略

数据本地化优化

合理设置InputSplit大小：InputSplit大小默认与HDFS块大小一致（128MB），可根据数据特性和集群情况调整。过小的Split会增加任务调度开销，过大的Split可能导致数据本地化率降低。
避免数据倾斜：数据倾斜指某些Key的出现频率远高于其他Key，导致处理这些Key的Reduce任务耗时过长。可通过以下方法解决：
- 使用自定义Partitioner均衡分配数据
- 对倾斜Key进行预处理，如拆分或合并
- 使用Map端Join减少Reduce端数据量

Shuffle阶段优化

合理设置缓冲区大小：Map任务的中间结果先存储在内存缓冲区中，当达到一定阈值（默认0.8）时溢写到磁盘。可通过mapreduce.task.io.sort.mb（默认100MB）调整缓冲区大小，增大缓冲区可减少溢写次数。
调整合并因子：合并溢写文件时，可通过mapreduce.task.io.sort.factor（默认10）调整合并因子，即一次合并的文件数量。增大合并因子可减少合并轮次。
启用压缩：对Map输出的中间结果进行压缩，可减少Shuffle阶段的数据传输量。常用的压缩算法有Snappy、LZO等，可通过mapreduce.map.output.compress=true启用压缩，并通过mapreduce.map.output.compress.codec指定压缩算法。

任务配置优化

合理设置Map和Reduce任务数量：
- Map任务数量：通常设置为输入数据大小 / InputSplit大小，一般为集群CPU核心数的1-2倍。
- Reduce任务数量：根据作业需求和数据量设置，一般为集群CPU核心数的0.9-1.75倍。过多的Reduce任务会增加任务调度和合并开销，过少的Reduce任务可能导致负载不均衡。
调整JVM参数：为Map和Reduce任务分配足够的内存，避免OOM（Out Of Memory）错误。可通过mapreduce.map.memory.mb和mapreduce.reduce.memory.mb设置任务内存大小，通过mapreduce.map.java.opts和mapreduce.reduce.java.opts设置JVM参数。
使用Combiner：Combiner可在Map端对中间结果进行本地聚合，减少Shuffle阶段的数据传输量。但需注意Combiner的输出必须与Reduce函数的输入兼容，且Combiner的执行是不确定的，不能依赖其执行次数。

资源管理优化

合理分配资源：根据集群资源状况和作业需求，为Map和Reduce任务分配合适的CPU和内存资源。可通过YARN的资源调度策略（如Capacity Scheduler或Fair Scheduler）实现资源的合理分配。
启用推测执行：当某个任务执行速度明显慢于同类型其他任务时，推测执行机制会启动一个备份任务，执行相同的任务，取先完成的任务结果。可通过mapreduce.map.speculative和mapreduce.reduce.speculative启用推测执行，但对于耗时较长或非幂等的任务，应禁用推测执行。