Rockmaple Blog

Hadoop the definite guide 读书笔记（五）

MapReduce的类型和格式

MapReduce类型

hadoop里的MapReduce有如下通用的格式：

map: (K1, V1) → list(K2, V2)
reduce: (K2, list(V2)) → list(K3, V3)

通常情况下，输入的key和value的类型(K1和V1)和输出的类型(K2和V2)是不同的，当然，Reduce的输入和Map的输出类型必须是一样的，虽然Reduce的输出可能是另外一种类型(K3和V3)。Java接口定义：

public interface Mapper<K1, V1, K2, V2> extends JobConfigurable, Closeable {
    void map(K1 key, V1 value, OutputCollector<K2, V2> output, Reporter reporter)
    throws IOException;
}

public interface Reducer<K2, V2, K3, V3> extends JobConfigurable, Closeable {
    void reduce(K2 key, Iterator<V2> values,
    OutputCollector<K3, V3> output, Reporter reporter) throws IOException;
}

如果使用combine函数的话，它同reduce函数的形式是一样的(同时也实现Reducer接口)，不同之处在于输出的类型是一些中间的key和value:

map: (K1, V1) → list(K2, V2)
combine: (K2, list(V2)) → list(K2, V2)
reduce: (K2, list(V2)) → list(K3, V3)

partitation函数处理中间的key和value的类型，并且返回其partation index：

 public interface Partitioner<K2, V2> extends JobConfigurable {
   int getPartition(K2 key, V2 value, int numPartitions);
 }

Input Splits 和 Records

Input会被分成input split，split由record 组成。map处理每一个record，并且返回key和value的对。

InputSplit由InputSplit接口来定义:

public interface InputSplit extends Writable {
    long getLength() throws IOException;
    String[] getLocations() throws IOException;
}

MapReduce程序作者并不需要直接处理InputSplit，因为它是由InputFormat创建的：

public interface InputFormat<K, V> {
    InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;
    RecordReader<K, V> getRecordReader(InputSplit split,
    JobConf job,
    Reporter reporter) throws IOException;
}

InputFormat类结构图：

OutputFormat

OutputFormat和InputFormat的类型相对应

Hadoop the definite guide 读书笔记（四）

MapReduce是如何工作的

MapReduce的整个工作过程如下图所示：

整个模型的最上层有四个实体：

• 客户端，负责提交MapReduce Job
• jobtracker， 负责调节job运行，jobtracker是一个java应用程序，main class名为JobTracker
• tasktrackers，job被分成多个split，tasktracker负责运行split，tasktracker是一个java应用程序，main class名为TaskTracker
• HDFS，用于共享job所需的文件

Job提交

JobClient调用submitJob方法提交一个job，一旦Job被提交，runJob方法会每隔一秒检查一下job的进度，如果进度有变化则显示在控制台。job完成后，如果job成功，在控制台显示job计数器，失败则在控制台显示错误原因。

JobClient的submitJob方法会完成以下工作：

• 从jobtracker那里拿到一个新job ID(通过调用JobTracker的getNewJobId()方法)
• 检查job的输出目录是否正常，比如，如果没有指定输出目录或者输出目录已经存在，则不提交job并抛出异常
• 计算job的input split，如果input split不能被计算出来，则抛出异常
• 拷贝job运行所需要的必要资源到jobtracker的文件系统里以job ID命名的目录下，这些资源包括jar文件，配置文件和input splits
• 告诉jobtracker可以运行job了

Job初始化

jobtracker接到submitJob的调用后，会将其放入一个内部的队列中，job scheduler会从队列里将job取出并初始化。

为了创建一个可以运行的task的队列，job scheduler首先从共享的文件系统中将input split取出来，并为每一个split创建一个map task，Reduce task的数据由mapred.reduce.tasks决定

Task分配

tasktracker定期向jobtracker发送heartbeat信息，做为heartbeat的一部分，tasktracker会表明其是否能可以接受新任务，如果可以，jobtracker通过heartbeat的返回值为其分配一个新的任务。

在为tasktracker分配task之前，jobtracker必须先选择一个job，有多种调度算法，简单的是保存一个job的优先级队列。一旦选择了一个job，jobtracker就选择一个task。

Task 执行

现在tasktracker已经被分配了一个task，下一步就是运行这个task。首先，从共享文件系统里拷贝job的jar文件至本地，其次，为task创建一个本地工作目录，把jar包解压至该目录下，最后，创建一个TaskRunner来运行这个task。

TaskRunner启动一个新的java虚拟机来运行每一个task。当然，也可以在不同的task之间共享JVM

进度和状态更新

MapReduce job是长期运行的批处理作业，因为运行时间很长，用户需要了解进度信息。Job以及其所属的task都有status信息，在整个job的运行过程中，status信息都会发生变化。

task运行时会记录其进度信息，并定期通过heartbeat向tasktracker报告。

Job完成

当jobtracker接收到信息，最后一个job已经完成，它会将job的状态设为”success”，这样，当JobClient查询job的状态时，它知道job已经成功完成，并通知用户，同时从runJob()方法中退出。