MapReduce编程模型

1、MapReduce编程模型 概述 MapReduce是在总结大量应用的共同特点的基础上抽象出来的分布式框 架，其适用应用场景的共同特点是：任务可被分解为相互独立的子问题。 基于该特点， MapReduce编程模型给出了其分布式编程方法，共分5个步 骤： 迭代迭代（iteration）。遍历输入数据，并将之解析成key/value对。 将输入的key/value对映射映射（map）成另外一些key/value对。 依据key对中间数据进行分组分组（group）。 以组为单位对数据进行规约规约（reduce）。 迭代迭代。将最终产生的key/value对保存到输出文件。 好处：组件化和并行化。 M

2、apReduce编程接口体系结构 整个编程模型位于应用程序层和MapReduce执行器之间，可以分为两层。 第一层是最基本的Java API，主要有5个可编程组件，分别是 InputFormat、Mapper、Partitioner、Reducer和OutputFormat。 Hadoop自带了很多直接可用的InputFormat、Partitioner和OutputFormat， 大部分情况下，用户只需编写Mapper和Reducer即可。第二层是工具层， 位于基本Java API之上，主要是为了方便用户编写复杂的MapReduce程 序和利用其他编程语言增加MapReduce计算平台的兼容

3、性而提出来的。 新旧MapReduce API比较 从0.20.0版本开始，Hadoop同时提供了新旧两套MapReduce API。新API 在旧API基础上进行封装，使其在扩展性和易用性方面更好。二者区别如 下： 存放位置 -旧API放在org.apache.Hadoop.mapred包中； -新API放在org.apache.Hadoop.mapreduce包及其子包中。 接口变为抽象类 -接口，严格的“协议约束”，只有方法声明而没有方法实，要 求所有实现类（抽象类除外）必须实现接口中的每个方法。 -抽象类，较宽松的“约束协议”，可为某些方法提供默认实 现，而继承类则可选择是否重新实现这

4、些方法。故而抽象类在类 衍化方面更有优势，即具有良好的向后兼容性。 上下文封装 -新版API将变量和函数封装成各种上下文（Context）类，使得 API具有更好的易用性和扩展性。 MapReduce API基本概念 序列化序列化 序列化是指将结构化对象转化为字节流以便于通过网络进行传输或 写入持久存储的过程。在Hadoop MapReduce中，序列化的主要作 用是永久存储和进程间通信。 为能够读取或者存储Java对象，MapReduce编程模型要求用户输入 和输出数据中的key和value必须是可序列化的。使一个Java对象可 序列化的方法是让其对应的实现Writable接口。而作为数据排

5、序的关 键字key所对应的类需要实现WritableComparable接口。 MapReduce API基本概念 Reporter参数参数 Reporter是MapReduce提供给应用程序的工具。如图3-4所示，应用程 序可使用Reporter中的方法报告完成进度（progress）、设定状态消 息（setStatus）以及更新计数器（incrCounter）。 Reporter是一个基础参数。MapReduce对外提供的大部分组件，包括 InputFormat、Mapper和Reducer等，均在其主要方法中添加了该参数。 MapReduce API基本概念 回调机制回调机制 回调机制是

6、一种常见的设计模式。它将工作流内的某个功能按照约 定的接口暴露给外部使用者，为外部使用者提供数据，或要求外部 使用者提供数据。 Hadoop MapReduce对外提供的5个组件（InputFormat、Mapper、 Partitioner、Reducer和OutputFormat）实际上全部属于回调接口。 当用户按照约定实现这几个接口后，MapReduce运行时环境会自动 调用它们。 MapReduce API基本概念 回调机制回调机制 如图3-5所示，MapReduce给用户暴露了接口Mapper，当用户按照 自己的应用程序逻辑实现自己的MyMapper后，Hadoop MapReduc

7、e 运行时环境会将输入数据解析成key/value对，并调用map()函数迭代 处理。 Java API解析 作业配置与提交 InputFormat接口的设计与实现 OutputFormat接口的设计与实现 Mapper与Reducer的解析 Partitioner接口的设计与实现 作业配置与提交 Hadoop配置文件介绍 在Hadoop中，Common、HDFS和MapReduce各有对应的配置文 件，用于保存对应模块中可配置的参数。 系统默认配置文件分别是core-default.xml、hdfs-default.xml和 mapred-default.xml，它们包含了所有可配置属性的默

8、认值。 而管理员自定义配置文件分别是core-site.xml、hdfs-site.xml和 mapred-site.xml。它们由管理员设置，主要用于定义一些新的配置 属性或者覆盖系统默认配置文件中的默认值。通常这些配置一旦确 定，便不能被修改（如果想修改，需重新启动Hadoop）。 需要注意的是，core-default.xml和core-site.xml属于公共基础库的 配置文件，默认情况下，Hadoop总会优先加载它们。 在Hadoop中，每个配置属性主要包括三个配置参数：name、value 和description，分别表示属性名、属性值和属性描述。 Hadoop为配置文件添加了两

9、个新的特性：final参数和变量扩展。 作业配置与提交 MapReduce作业配置与提交 在MapReduce中，每个作业由两部分组成：应用程序和 作业配置。 其中，作业配置内容包括环境配置和用户自定义配置两 部分。 环境配置由Hadoop自动添加，主要由mapred- default.xml和mapred-site.xml两个文件中的配置选项组 合而成；用户自定义配置则由用户自己根据作业特点个 性化定制而成，比如用户可设置作业名称，以及 Mapper/Reducer、Reduce Task个数等。 作业配置与提交 MapReduce作业配置与提交 新版API用Job类代替了JobConf和J

10、obClient两个类，这样，仅使用 一个类的同时可完成作业配置和作业提交相关功能，进一步简化了 作业编写方式。 作业配置与提交 旧API中的作业配置 MapReduce配置模块代码结构 如图3-6所示。其中， org.apache.hadoop.conf中的 Configuration类是配置模块最 底层的类。从图3-6中可以看出， 该类支持以下两种基本操作。 序列化：序列化是将结构化数 据转换成字节流，以便于传输 或存储。Java实现了自己的一 套序列化框架。凡是需要支持 序列化的类，均需要实现 Writable接口。 迭代：为了方便遍历所有属性， 它实现了Java开发包中的 Iterat

11、or接口。 作业配置与提交 旧API中的作业配置 Configuration类总会依次加载 core-default.xml和core- site.xml两个基础配置文件。 JobConf类描述了一个 MapReduce作业运行时需要的 所有信息，而MapReduce运行 时环境正是根据JobConf提供 的信息运行作业的。 JobConf继承了Configuration 类，并添加了一些设置/获取作 业属性的setter/getter函数，以 方便用户编写MapReduce程序。 默认情况下，JobConf会自动 加载配置文件mapred- default.xml和mapred-site.x

12、ml 作业配置与提交 新API中的作业配置 与新API中的作业配置相关的类是Job，该类同时具有作业配置和作 业提交的功能。作业配置部分的类图如图3-7所示。Job类继承了一 个新类JobContext，而Context自身则包含一个JobConf类型的成员。 注意，JobContext类仅提供了一些getter方法，而Job类中则提供了 一些setter方法。 InputFormat接口的设计与实现 InputFormat主要用于描述输入数据的格式，它提供 以下两个功能。 数据切分：按照某个策略将输入数据切分成若干 个split，以便确定Map Task个数以及对应的split。 为Mapp

13、er提供输入数据：给定某个split，能将其 解析成一个个key/value对。 InputFormat接口的设计与实现 旧版API的InputFormat解析 如图3-8所示，在旧API中InputFormat是一个接口，包含两种方法。 getSplits方法主要完成数据切分的功能，它会尝试着将输入数据切分 成numSplits个InputSplit。 getRecordReader方法返回一个RecordReader对象，该对象可将输 入的InputSplit解析成若干个key/value对。MapReduce框架在Map Task执行过程中，会不断调用RecordReader对象中的方法

14、，迭代获 取key/value对并交给map()函数处理。 InputFormat接口的设计与实现 InputFormat的实现 所有基于文件的InputFormat实现的基类是FileInputFormat，它最重要的 功能是为各种InputFormat提供统一的getSplits函数。该函数实现中最 核心的两个算法是文件切分算法和host选择算法。 （1）文件切分算法）文件切分算法 文件切分算法主要用于确定InputSplit的个数以及每个InputSplit对应的数 据段。FileInputFormat以文件为单位切分生成InputSplit。对于每个文 件，由以下三个属性值确定其对应的

15、InputSplit的个数。 goalSize：它是根据用户期望的InputSplit数目计算出来的，即 totalSize/numSplits。其中，totalSize为文件总大小；numSplits为用 户设定的Map Task个数，默认情况下是1。 minSize：InputSplit的最小值，由配置参数mapred.min.split.size确定， 默认是1。 blockSize：文件在HDFS中存储的block大小，不同文件可能不同， 默认是64 MB。 三个参数共同决定InputSplit的最终大小，计算方法： splitSize = maxminSize, mingoalSiz

16、e, blockSize InputFormat接口的设计与实现 （2）host选择算法选择算法 待InputSplit切分方案确定后，下一步要确定每个InputSplit的元数据信息。 这通常由四部分组成：，分别表示InputSplit所 在的文件、起始位置、长度以及所在的host（节点）列表。其中，前三 项很容易确定，难点在于host列表的选择方法。 InputSplit的host列表选择策略直接影响到运行过程中的任务本地性。 HDFS上的文件是以block为单位组织的，一个大文件对应的block可能遍 布整个Hadoop集群，而InputSplit的划分算法可能导致一个InputSpli

17、t对 应多个block ，这些block可能位于不同节点上，这使得Hadoop不可能实 现完全的数据本地性。 InputSplit对应的block可能位于多个节点上，但考虑到任务调度的效率， 通常不会把所有节点加到InputSplit的host列表中，而是选择包含（该 InputSplit）数据总量最大的前几个节点（Hadoop限制最多选择10个， 多余的会过滤掉），以作为任务调度时判断任务是否具有本地性的主要 凭证。为此，FileInputFormat设计了一个简单有效的启发式算法：首先 按照rack包含的数据量对rack进行排序，然后在rack内部按照每个node 包含的数据量对node排

18、序，最后取前N个node的host作为InputSplit的 host列表，这里的N为block副本数。这样，当任务调度器调度Task时， 只要将Task调度给位于host列表的节点，就认为该Task满足本地性。 InputFormat接口的设计与实现 【实例】某个Hadoop集群的网络拓扑结构如图3-10所示，HDFS中block 副本数为3，某个InputSplit包含3个block，大小依次是100、150和75， 很容易计算，4个rack包含的（该InputSplit的）数据量分别是175、250、 150和75。rack2中的node3和node4， rack1中的node1将被添加

19、到该 InputSplit的host列表中。 从以上host选择算法可知，当InputSplit尺寸大于block尺寸时，Map Task 并不能实现完全数据本地性，也就是说，总有一部分数据需要从远程节 点上读取，因而当使用基于FileInputFormat实现InputFormat时，为了提 高Map Task的数据本地性，应尽量使InputSplit大小与block大小相同。 InputFormat接口的设计与实现 新版API的InputFormat解析 新版API的InputFormat类图如图3-11所示。新API与旧API比较，在 形式上发生了较大变化，但仔细分析，发现仅仅是对之前的

20、一些类 进行了封装。通过封装，使接口的易用性和扩展性得以增强。 InputFormat接口的设计与实现 新版API的InputFormat解析 此外，对于基类FileInputFormat，新API中有一个值得注意的改动： InputSplit划分算法不再考虑用户设定的Map Task个数，而用 mapred.max.split.size（记为maxSize）代替，即InputSplit大小的计 算公式变为： splitSize = maxminSize, minmaxSize, blockSize OutputFormat接口的设计与实现 OutputFormat主要用于描述输出数据的格式，

21、它能够将用户提供的 key/value对写入特定格式的文件中。 旧版API的OutputFormat解析 如图3-12所示，在旧版API中，OutputFormat是一个接口，它包含两个 方法。 checkOutputSpecs方法一般在用户作业被提交到JobTracker之前，由 JobClient自动调用，以检查输出目录是否合法。 getRecordWriter方法返回一个RecordWriter类对象。该类中的方法write 接收一个key/value对，并将之写入文件。在Task执行过程中， MapReduce框架会将map()或者reduce()函数产生的结果传入write方法。 O

22、utputFormat接口的设计与实现 OutputFormat的实现 基类FileOutputFormat需要提供所有基于文件的OutputFormat实现的公 共功能，总结起来，主要有以下两个： （1）实现checkOutputSpecs接口 该接口在作业运行之前被调用，默认功能是检查用户配置的输出目录是 否存在，如果存在则抛出异常，以防止之前的数据被覆盖。 （2）处理side-effect file 任务的side-effect file并不是任务的最终输出文件，而是具有特殊用途的 任务专属文件。它的典型应用是执行推测式任务。在Hadoop中，因为硬 件老化、网络故障等原因，同一个作业的

23、某些任务执行速度可能明显慢 于其他任务，这种任务会拖慢整个作业的执行速度。为了对这种“慢任 务”进行优化，Hadoop会为之在另外一个节点上启动一个相同的任务， 该任务便被称为推测式任务，最先完成任务的计算结果便是这块数据对 应的处理结果。为防止这两个任务同时往一个输出文件中写入数据时发 生写冲突，FileOutputFormat会为每个Task的数据创建一个side-effect file，并将产生的数据临时写入该文件，待Task完成后，再移动到最终输 出目录中。 OutputFormat接口的设计与实现 Side-effect file的相关操作，比如创建、删除、移动等，均由 Output

24、Committer完成。它是一个接口， Hadoop提供了默认实现 F i l e O u t p u t C o m m i t t e r ， 用 户 也 可 以 根 据 自 己 的 需 求 编 写 OutputCommitter实现，并通过参数mitter.class指 定。OutputCommitter接口定义以及FileOutputCommitter对应的实现如 表3-2所示。 表3-2OutputCommitter接口定义以及FileOutputCommitter对应的实现 OutputFormat接口的设计与实现 新版API的OutputFormat解析 如图3-14所示，除了接

25、口变为抽象类外，新API中的OutputFormat增 加了一个新的方法：getOutputCommitter，以允许用户自己定制合 适的OutputCommitter实现。 Mapper与Reducer解析 旧版API的Mapper/Reducer解析 Mapper/Reducer中封装了应用程序的数据处理逻辑。为简化接口， MapReduce要求所有存储在底层分布式文件系统上的数据均要解释 成key/value的形式，并交给Mapper/Reducer中的map/reduce函数 处理，产生另外一些key/value。 Mapper类图如图3-15所示，包括初始化、Map操作和清理三部分。

26、 Mapper与Reducer解析 （1）初始化）初始化 Mapper继承了JobConfigurable接口。该接口中的configure方法允许通 过JobConf参数对Mapper进行初始化。 （2）Map操作操作 MapReduce框架会通过InputFormat中RecordReader从InputSplit获取一 个个key/value对，并交给下面的map()函数处理： void map(K1 key, V1 value, OutputCollector output, Reporter reporter) throws IOException; 该函数的参数除了key和valu

27、e之外，还包括OutputCollector和Reporter 两个类型的参数，分别用于输出结果和修改Counter值。 （3）清理）清理 Mapper通过继承Closeable接口（它又继承了Java IO中的Closeable接 口）获得close方法，用户可通过实现该方法对Mapper进行清理。 Mapper与Reducer解析 MapReduce提供了很多Mapper/Reducer实现，但大部分功能比较简单， 具体如图3-16所示。它们对应的功能分别是： ChainMapper/ChainReducer：用于支持链式作业。 IdentityMapper/IdentityReducer

28、：对于输入key/value不进行任何处理， 直接输出。 InvertMapper：交换key/value位置。 RegexMapper：正则表达式字符串匹配。 TokenMapper：将字符串分割成若干个token（单词），可用作 WordCount的Mapper。 LongSumReducer：以key为组，对long类型的value求累加和。 Mapper与Reducer解析 新版API的Mapper/Reducer解析 新API在旧API基础上发生了以下几个变化： Mapper由接口变为抽象类，且不再继承JobConfigurable和 Closeable两个接口，而是直接在类中添加了

29、setup和cleanup两个方 法进行初始化和清理工作。 将参数封装到Context对象中，这使得接口具有良好的扩展性。 去掉MapRunnable接口，在Mapper中添加run方法，以方便用户定 制map()函数的调用方法，run默认实现与旧版本中MapRunner的run 实现一样。 新API中Reducer遍历value的迭代器类型变为java.lang.Iterable，使 得用户可以采用“foreach”形式遍历所有value，如下所示： Mapper与Reducer解析 新版API的Mapper/Reducer解析 Partitioner接口的设计与实现 Partitioner

30、的作用是对Mapper产生的中间结果进行分片，以便将同一分 组的数据交给同一个Reducer处理，它直接影响Reduce阶段的负载均衡。 旧版API中Partitioner的类图如图3-20所示。它继承了JobConfigurable， 可通过configure方法初始化。它本身只包含一个待实现的方法 getPartition。该方法包含三个参数，均由框架自动传入，前面两个参数 是key/value，第三个参数numPartitions表示每个Mapper的分片数，也 就是Reducer的个数。 Partitioner接口的设计与实现 MapReduce提供了两个Partitioner实现：H

31、ashPartitioner和 TotalOrderPartitioner。其中HashPartitioner是默认实现，它实现了一种 基于哈希值的分片方法，代码如下： Partitioner接口的设计与实现 TotalOrderPartitioner提供了一种基于区间的分片方法，通常用在数据全 排序中。在MapReduce环境中，容易想到的全排序方案是归并排序，即 在Map阶段，每个Map Task进行局部排序；在Reduce阶段，启动一个 Reduce Task进行全局排序。由于作业只能有一个Reduce Task，因而 Reduce阶段会成为作业的瓶颈。为了提高全局排序的性能和扩展性，

32、MapReduce 提供了TotalOrderPartitioner。它能够按照大小将数据分成 若干个区间（分片），并保证后一个区间的所有数据均大于前一个区间 数据，这使得全排序的步骤如下： 步骤步骤1数据采样。在Client端通过采样获取分片的分割点。Hadoop自带 了几个采样算法，如IntercalSampler、RandomSampler、SplitSampler 等（具体见org.apache.hadoop.mapred.lib包中的InputSampler类）。下 面举例说明。 采样数据为：b，abc，abd，bcd，abcd，efg，hii，afd，rrr，mnk 经排序后得到：abc，abcd，abd，afd，b，bc