3.3 MapReduce 作业运行与管理

3.3 MapReduce 作业运行与管理

3.3 MapReduce 作业运行与管理

3.3.1 作业提交与初始化

MapReduce 作业的生命周期始于客户端的提交。用户编写 MapReduce 程序后，需要通过客户端将作业提交到 Hadoop 集群。

1. 作业提交过程

客户端提交作业主要涉及以下几个步骤：

• 编写 MapReduce 程序: 用户根据业务需求，使用 Hadoop MapReduce API 编写 Map 函数和 Reduce 函数，以及驱动程序 (Driver)。

• 配置作业: 在驱动程序中，用户需要配置作业的各种参数，例如输入路径、输出路径、Map 和 Reduce 类、数据格式、分区器、排序器等。这些配置信息将指导 MapReduce 框架如何执行作业。

• 作业提交: 客户端通过 JobClient.submitJob() (Hadoop 1.x) 或 Job.submit() (Hadoop 2.x) 方法将作业提交到 ResourceManager (Hadoop 2.x) 或 JobTracker (Hadoop 1.x)。

2. 作业初始化

当 ResourceManager (或 JobTracker) 接收到作业提交请求后，会进行一系列初始化操作：

• 资源请求: ResourceManager (或 JobTracker) 接收到客户端提交的作业后，会创建一个新的 ApplicationMaster 实例 (Hadoop 2.x)。ApplicationMaster 负责整个作业的生命周期管理和资源调度。在 Hadoop 1.x 中，JobTracker 承担了类似 ApplicationMaster 的角色。

• 作业规划: ApplicationMaster (或 JobTracker) 读取作业配置信息，分析输入数据，计算输入分片 (InputSplit) 的数量。每个输入分片将对应一个 Map Task。同时，根据配置的 Reduce Task 数量，确定 Reduce Task 的数量。

• 资源分配: ApplicationMaster 向 ResourceManager 请求运行 Map Task 和 Reduce Task 所需的资源 (Container)。ResourceManager 根据集群资源情况，为 ApplicationMaster 分配 Container 资源。

• 作业元数据存储: 作业的配置信息、输入分片信息、任务数量等元数据会被存储在 ResourceManager (或 JobTracker) 的内存中，并可能持久化到磁盘或 ZooKeeper 中，以便进行作业监控和管理。

Mermaid 图示：作业提交与初始化流程

代码实践：作业配置和提交 (Hadoop 2.x Java API)

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;
public class WordCount {
    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 {
            String[] words = value.toString().toLowerCase().split("\\s+"); // 简单分词
            for (String w : words) {
                word.set(w);
                context.write(word, one);
            }
        }
    }
    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);
        }
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count"); // 创建 Job 实例
        job.setJarByClass(WordCount.class); // 设置主类
        job.setMapperClass(TokenizerMapper.class); // 设置 Mapper 类
        job.setCombinerClass(IntSumReducer.class); // 设置 Combiner 类 (可选)
        job.setReducerClass(IntSumReducer.class); // 设置 Reducer 类
        job.setOutputKeyClass(Text.class); // 设置输出 Key 类型
        job.setOutputValueClass(IntWritable.class); // 设置输出 Value 类型
        FileInputFormat.addInputPath(job, new Path(args[0])); // 设置输入路径
        FileOutputFormat.setOutputPath(job, new Path(args[1])); // 设置输出路径
        System.exit(job.waitForCompletion(true) ? 0 : 1); // 提交作业并等待完成
    }
}
​

代码详解:

• Configuration conf = new Configuration();: 创建 Hadoop 配置对象，加载 Hadoop 配置文件 (core-site.xml, hdfs-site.xml, mapred-site.xml, yarn-site.xml)。

• Job job = Job.getInstance(conf, "word count");: 创建 Job 实例，并设置作业名称为 "word count"。

• job.setJarByClass(WordCount.class);: 设置包含 Mapper 和 Reducer 类的 JAR 包，通常设置为驱动程序类所在的类。

• job.setMapperClass(TokenizerMapper.class);, job.setReducerClass(IntSumReducer.class);: 分别设置 Mapper 类和 Reducer 类。

• job.setCombinerClass(IntSumReducer.class);: 设置 Combiner 类，用于在 Map 端本地聚合数据，减少网络传输 (可选)。

• job.setOutputKeyClass(Text.class);, job.setOutputValueClass(IntWritable.class);: 设置作业最终输出的 Key 和 Value 类型。

• FileInputFormat.addInputPath(job, new Path(args[0]));, FileOutputFormat.setOutputPath(job, new Path(args[1]));: 设置输入路径和输出路径，路径从命令行参数传入。

• System.exit(job.waitForCompletion(true) ? 0 : 1);: 提交作业到 Hadoop 集群，并等待作业完成。waitForCompletion(true) 会打印作业进度信息。

3.3.2 任务调度与执行

作业初始化完成后，进入任务调度和执行阶段。MapReduce 作业主要包含两种类型的任务：Map Task 和 Reduce Task。

1. 任务调度

• 数据本地性: 任务调度器 (ApplicationMaster 或 JobTracker) 会尽量将 Map Task 调度到存储输入数据分片的 DataNode 上，以实现数据本地性，减少数据传输开销。

• 资源调度: 任务调度器会根据集群资源情况和任务优先级，为任务分配 Container 资源。

• 任务分配: 任务调度器将 Map Task 和 Reduce Task 分配给 NodeManager (或 TaskTracker) 上的 Container 执行。

2. Map Task 执行

Map Task 的执行过程主要包括以下步骤：

• Input Split: 每个 Map Task 处理一个输入分片 (InputSplit)，InputFormat 负责将输入数据划分为 InputSplit。

• Map 函数: Map Task 调用用户自定义的 Map 函数，处理输入分片中的每条记录，生成中间结果 (Key-Value 对)。

• Shuffle (Map 端): Map Task 的中间结果首先被写入内存缓冲区，当缓冲区达到一定阈值或记录数时，会溢写到磁盘。在溢写之前，可以进行可选的 Combiner 操作，对中间结果进行本地聚合。溢写到磁盘的文件会根据 Reduce Task 的数量进行分区 (Partition)，并按照 Key 进行排序 (Sort)。

• 磁盘合并: 当 Map Task 完成所有数据处理后，磁盘上可能会存在多个溢写文件。这些文件会被合并成一个或多个更大的文件，并进行再次排序。

3. Reduce Task 执行

Reduce Task 的执行过程主要包括以下步骤：

• Shuffle (Reduce 端): Reduce Task 从各个 Map Task 的输出结果中，拉取 (Fetch) 属于自己分区的数据。拉取的数据会先存储在内存缓冲区，当缓冲区满或达到阈值时，会溢写到磁盘。

• 合并排序 (Merge Sort): Reduce Task 将从不同 Map Task 拉取的数据进行合并和排序，形成一个有序的数据集。

• Reduce 函数: Reduce Task 调用用户自定义的 Reduce 函数，对排序后的数据集进行处理，生成最终结果。

• Output: Reduce Task 将最终结果写入到 HDFS 的输出路径。OutputFormat 负责将 Reduce 函数的输出结果格式化并写入到输出文件。

Mermaid 图示：MapReduce 任务执行流程

代码实践：自定义 Partitioner (分区器)

默认情况下，MapReduce 使用 HashPartitioner 根据 Key 的哈希值进行分区。用户可以自定义 Partitioner，实现更灵活的分区策略。

import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.IntWritable;
public class CustomPartitioner extends Partitioner<Text, IntWritable> {
    @Override
    public int getPartition(Text key, IntWritable value, int numPartitions) {
        String word = key.toString();
        if (word.startsWith("a")) {
            return 0 % numPartitions; // 以 'a' 开头的单词分配到分区 0
        } else if (word.startsWith("b")) {
            return 1 % numPartitions; // 以 'b' 开头的单词分配到分区 1
        } else {
            return 2 % numPartitions; // 其他单词分配到分区 2
        }
        // 注意：实际应用中需要更严谨的逻辑，并处理 numPartitions 为 0 的情况
    }
}
​

代码详解:

• 自定义 Partitioner 需要继承 org.apache.hadoop.mapreduce.Partitioner 类。

• getPartition(Text key, IntWritable value, int numPartitions) 方法决定了 Key-Value 对应该被分配到哪个分区。

• numPartitions 参数表示 Reduce Task 的数量，也是分区的数量。

• 在驱动程序中，需要设置自定义 Partitioner：

job.setPartitionerClass(CustomPartitioner.class);
job.setNumReduceTasks(3); // 设置 Reduce Task 数量，与分区数量一致
​

3.3.3 作业监控与管理

MapReduce 框架提供了丰富的机制来监控和管理正在运行的作业，以及已完成的作业。

1. 作业监控

• Web UI: Hadoop ResourceManager (或 JobTracker) 和 ApplicationMaster 都提供了 Web UI，用户可以通过 Web 浏览器访问这些 UI，查看作业的运行状态、进度、任务列表、日志、计数器等信息。

  • ResourceManager Web UI (Hadoop 2.x): 通常端口为 8088。可以查看集群资源使用情况、应用程序列表、ApplicationMaster 信息等。

  • ApplicationMaster Web UI (Hadoop 2.x): 每个 ApplicationMaster 都有独立的 Web UI，端口在 ResourceManager UI 中可以找到。可以查看作业的详细信息，包括 Map Task 和 Reduce Task 的进度、状态、日志、计数器等。

  • JobTracker Web UI (Hadoop 1.x): 通常端口为 50030。可以查看作业列表、作业详细信息、任务列表、日志、计数器等。

• 命令行工具: Hadoop 提供了 yarn (Hadoop 2.x) 和 hadoop job (Hadoop 1.x) 等命令行工具，用于查询作业状态、日志、计数器等信息。

  • yarn application -list: 列出正在运行的应用程序 (包括 MapReduce 作业)。

  • yarn application -status <Application ID>: 查看指定应用程序的状态信息。

  • yarn logs -applicationId <Application ID>: 查看应用程序的日志。

  • hadoop job -status <Job ID> (Hadoop 1.x): 查看指定作业的状态信息。

  • hadoop job -history <Job History 文件路径> (Hadoop 1.x): 查看已完成作业的历史信息。

• 计数器 (Counters): MapReduce 框架提供了计数器机制，用于统计作业执行过程中的各种指标，例如输入记录数、输出记录数、Map Task 数量、Reduce Task 数量、文件读写字节数等。用户也可以自定义计数器，统计特定业务指标。计数器信息可以在 Web UI 和命令行工具中查看。

2. 作业管理

• 作业控制: 用户可以通过命令行工具或 API 对正在运行的作业进行控制，例如 Kill 作业。

  • yarn application -kill <Application ID> (Hadoop 2.x): Kill 指定应用程序。

  • hadoop job -kill <Job ID> (Hadoop 1.x): Kill 指定作业。

• 作业优先级: Hadoop 允许设置作业的优先级，高优先级的作业可以优先获得资源调度。

• 资源配置: 用户可以在作业配置中设置 Map Task 和 Reduce Task 的资源需求 (例如内存、CPU)。ResourceManager (或 JobTracker) 会根据资源需求进行任务调度。

• 故障处理: MapReduce 框架具有一定的容错机制。当任务执行失败时，ResourceManager (或 JobTracker) 会尝试重新调度任务。

Mermaid 图示：作业监控与管理

代码实践：自定义计数器

import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Counters;
import java.io.IOException;
public class CounterMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
    public enum CUSTOM_COUNTERS {
        LINES_PROCESSED,
        INVALID_LINES
    }
    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        Counters counters = context.getCounter(Mapper.class.getName()); // 获取计数器组
        counters.findCounter(CUSTOM_COUNTERS.LINES_PROCESSED).increment(1); // 递增行处理计数器
        String line = value.toString();
        if (line.startsWith("#")) {
            counters.findCounter(CUSTOM_COUNTERS.INVALID_LINES).increment(1); // 递增无效行计数器
            return; // 忽略以 '#' 开头的行
        }
        // ... 其他 Map 处理逻辑 ...
        String[] words = line.split("\\s+");
        for (String word : words) {
            context.write(new Text(word), new LongWritable(1));
        }
    }
}
​

代码详解:

• 在 Mapper 或 Reducer 类中，定义 enum 类型来声明自定义计数器组和计数器名称。

• context.getCounter(Mapper.class.getName()) 获取计数器组，参数可以是任意字符串，通常使用类名。

• counters.findCounter(CUSTOM_COUNTERS.LINES_PROCESSED) 获取指定的计数器。

• counter.increment(1) 递增计数器的值。

• 自定义计数器信息可以在 Web UI 和命令行工具中查看，例如在 ApplicationMaster Web UI 的 "Counters" 选项卡中。

3.3.4 总结

理解 MapReduce 作业的运行机制，有助于用户更好地进行作业开发、性能调优和故障排查。合理配置作业参数、利用监控工具、自定义 Partitioner 和 Counter 等技术，可以提升 MapReduce 作业的执行效率和管理水平，从而更好地利用 Hadoop 集群进行大数据处理。

在后续章节中，我们将继续深入探讨 MapReduce 的高级特性和优化技巧，以及 Hadoop 生态系统中其他相关组件的应用。