MapReduce JobClient: 如何优化大数据处理流程？

MapReduce作业调度详解之JobClient

在编写MapReduce程序时，通常会使用以下代码进行配置和提交作业：

Job job = new Job(conf, "ClosedDataCube");
job.setJarByClass(ClosedDataCube.class);
job.setMapperClass(cubeMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
job.setCombinerClass(cubeCombiner.class);
job.setReducerClass(cubeReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);

作业类（Job）

Job类允许用户配置作业、提交作业、控制执行以及查询状态，所有设置方法在提交作业时开始执行，主要方法包括：

setJarByClass(Class<?> c)：指定作业的主类。

setMapperClass(Class<? extends Mapper> mapperClass)：设置Mapper类。

setMapOutputKeyClass(Class<?> keyClass)：设置Map输出的键类型。

setMapOutputValueClass(Class<?> valueClass)：设置Map输出的值类型。

setCombinerClass(Class<? extends Reducer> combinerClass)：设置Combiner类。

setReducerClass(Class<? extends Reducer> reducerClass)：设置Reducer类。

setOutputKeyClass(Class<?> keyClass)：设置Reduce输出的键类型。

setOutputValueClass(Class<?> valueClass)：设置Reduce输出的值类型。

setInputFormat(Class<? extends InputFormat> inputFormat)：设置输入格式。

setOutputFormat(Class<? extends OutputFormat> outputFormat)：设置输出格式。

提交作业

提交作业的方式有两种：

1、简单提交：通过调用submit()方法直接将作业提交到Hadoop集群的平台，客户端没有任何日志输出。

2、等待完成：通过调用waitForCompletion()方法提交作业并等待其完成，同时会时刻打印作业执行的进度信息，如果作业出现异常，也会立刻将异常信息打印出来。

JobClient与作业提交流程

JobClient是用户与JobTracker联系的基本接口，提供提交作业的基本接口、跟踪作业、获得作业的运行状态信息等，具体流程如下：

1、创建作业的JobID并提交三个文件：job.xml（作业配置）、job.jar（包含执行任务需要的各种类）、job.split（文件分块的相关信息）。

2、检查作业的输入、输出。

3、计算map数目：通过InputFormat的getSplits(job)方法获得作业的split并将split序列化封装为RawSplit，返回split数目即代表map数目。

4、向JobTracker的fs中写入job文件。

5、真正的提交一个作业：通过JobClient.submitJobInternal(conf)方法提交作业，并返回一个RunningJob对象用于查询作业信息。

在上述过程中，JobClient通过RPC模块实现与JobTracker的通信，大致过程包括：

创建作业的JobID：通过JobClient类的Proxy接口调用创建了JobTracker对象，并与master取得联系。

提交作业并获得JobStatus对象：JobTracker创建job成功后会给JobClient传回一个JobStatus对象，用于记录job的状态信息。

监控作业进度：JobClient根据JobStatus对象创建一个NetworkedJob的RunningJob对象，定时从JobTracker获得执行过程的统计数据来监控并打印到用户的控制台。

常见问题解答（FAQs）

1、问题：为什么需要Combiner？

解答：Combiner用于在Map阶段之后、Reduce阶段之前对Map输出进行局部汇总，从而减少网络传输的数据量，提高作业效率，它的原理是合并那些具有相同key的value值，以减少数据在网络上的传输量，减轻Reducer的负担。

2、问题：什么是TaskTracker，它的作用是什么？

解答：TaskTracker负责执行由JobTracker分配的任务，每个TaskTracker节点上只有一个TaskTracker，它通过周期性地发送心跳来告知JobTracker其当前的健康状态，并接收来自JobTracker的任务分配和指令，TaskTracker负责启动和管理具体的Map或Reduce任务，并将任务的执行状态反馈给JobTracker。