MapReduce工作机制和序列化

作者: 依天立业 | 来源:发表于2018-08-20 21:07 被阅读0次

MapReduce执行流程

image

MapReduce的执行步骤

1、Map任务处理

1.1 读取HDFS中的文件。每一行解析成一个<k,v>。每一个键值对调用一次map函数。 <0,hello you> <10,hello me>

1.2 覆盖map()，接收1.1产生的<k,v>，进行处理，转换为新的<k,v>输出。　　　　　　　　　　<hello,1> <you,1> <hello,1> <me,1>

1.3 对1.2输出的<k,v>进行分区。默认分为一个区。详见Partitioner

1.4 对不同分区中的数据进行排序（按照k）、分组。分组指的是相同key的value放到一个集合中。　排序后：<hello,1> <hello,1> <me,1> <you,1> 分组后：<hello,{1,1}><me,{1}><you,{1}>

1.5 （可选）对分组后的数据进行归约。详见Combiner

2、Reduce任务处理

2.1 多个map任务的输出，按照不同的分区，通过网络copy到不同的reduce节点上。详见shuffle过程分析

2.2 对多个map的输出进行合并、排序。覆盖reduce函数，接收的是分组后的数据，实现自己的业务逻辑，　<hello,2> <me,1> <you,1>

处理后，产生新的<k,v>输出。

2.3 对reduce输出的<k,v>写到HDFS中。

Java代码实现

注：要导入org.apache.hadoop.fs.FileUtil.java。

1、先创建一个hello文件，上传到HDFS中

image

2、然后再编写代码，实现文件中的单词个数统计（代码中被注释掉的代码，是可以省略的，不省略也行）

<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; word-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;"> 1 package mapreduce; 2
3 import java.net.URI; 4 import org.apache.hadoop.conf.Configuration; 5 import org.apache.hadoop.fs.FileSystem; 6 import org.apache.hadoop.fs.Path; 7 import org.apache.hadoop.io.LongWritable; 8 import org.apache.hadoop.io.Text; 9 import org.apache.hadoop.mapreduce.Job; 10 import org.apache.hadoop.mapreduce.Mapper; 11 import org.apache.hadoop.mapreduce.Reducer; 12 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 13 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; 14 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 15 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; 16
17 public class WordCountApp { 18 static final String INPUT_PATH = "hdfs://chaoren:9000/hello";
19 static final String OUT_PATH = "hdfs://chaoren:9000/out";
20
21 public static void main(String[] args) throws Exception { 22 Configuration conf = new Configuration(); 23 FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf); 24 Path outPath = new Path(OUT_PATH); 25 if (fileSystem.exists(outPath)) { 26 fileSystem.delete(outPath, true);
27 }
28
29 Job job = new Job(conf, WordCountApp.class.getSimpleName());
30
31 // 1.1指定读取的文件位于哪里
32 FileInputFormat.setInputPaths(job, INPUT_PATH);
33 // 指定如何对输入的文件进行格式化，把输入文件每一行解析成键值对 34 //job.setInputFormatClass(TextInputFormat.class);
35
36 // 1.2指定自定义的map类
37 job.setMapperClass(MyMapper.class);
38 // map输出的<k,v>类型。如果<k3,v3>的类型与<k2,v2>类型一致，则可以省略 39 //job.setOutputKeyClass(Text.class);
40 //job.setOutputValueClass(LongWritable.class);
41
42 // 1.3分区 43 //job.setPartitionerClass(org.apache.hadoop.mapreduce.lib.partition.HashPartitioner.class);
44 // 有一个reduce任务运行 45 //job.setNumReduceTasks(1);
46
47 // 1.4排序、分组 48
49 // 1.5归约 50
51 // 2.2指定自定义reduce类
52 job.setReducerClass(MyReducer.class);
53 // 指定reduce的输出类型
54 job.setOutputKeyClass(Text.class);
55 job.setOutputValueClass(LongWritable.class);
56
57 // 2.3指定写出到哪里
58 FileOutputFormat.setOutputPath(job, outPath);
59 // 指定输出文件的格式化类 60 //job.setOutputFormatClass(TextOutputFormat.class);
61
62 // 把job提交给jobtracker运行
63 job.waitForCompletion(true);
64 }
65
66 /**
67 *
68 * KEYIN 即K1 表示行的偏移量
69 * VALUEIN 即V1 表示行文本内容
70 * KEYOUT 即K2 表示行中出现的单词
71 * VALUEOUT 即V2 表示行中出现的单词的次数，固定值1
72 *
73 /
74 static class MyMapper extends
75 Mapper<LongWritable, Text, Text, LongWritable> { 76 protected void map(LongWritable k1, Text v1, Context context) 77 throws java.io.IOException, InterruptedException { 78 String[] splited = v1.toString().split("\t");
79 for (String word : splited) { 80 context.write(new Text(word), new LongWritable(1));
81 }
82 };
83 }
84
85 /*
86 * KEYIN 即K2 表示行中出现的单词
87 * VALUEIN 即V2 表示出现的单词的次数
88 * KEYOUT 即K3 表示行中出现的不同单词
89 * VALUEOUT 即V3 表示行中出现的不同单词的总次数
90 */
91 static class MyReducer extends
92 Reducer<Text, LongWritable, Text, LongWritable> { 93 protected void reduce(Text k2, java.lang.Iterable<LongWritable> v2s, 94 Context ctx) throws java.io.IOException, 95 InterruptedException {
96 long times = 0L;
97 for (LongWritable count : v2s) { 98 times += count.get(); 99 } 100 ctx.write(k2, new LongWritable(times)); 101 }; 102 } 103 }</pre>

3、运行成功后，可以在Linux中查看操作的结果

image

</div>

MapReduce中的序列化

<div class="mdContent">
hadoop序列化的特点：

序列化格式特点：

1.紧凑：高效使用存储空间。
2.快速：读写数据的额外开销小
3.可扩展：可透明地读取老格式的数据
4.互操作：支持多语言的交互

hadoop序列化与java序列化的最主要的区别是：在复杂类型的对象下，hadoop序列化不用像java对象类一样传输多层的父子关系，需要哪个属性就传输哪个属性值，大大的减少网络传输的开销。

hadoop序列化的作用: <div class="mdContent">

1.序列化的在分布式的环境的作用：进程之间的通信，节点通过网络之间的

2.hadoop节点之间数据传输

节点1：（序列化二进制数据） ------->(二进制流消息) 节点2:(反序列化二进制数据)

MR中key,value都是需要实现WritableComparable接口的对象，这样的对象才是hadoop序列化的对象。

package com.feihao;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

public class StudentWritable implements WritableComparable<StudentWritable> {
private String name;
private int age;

public void write(DataOutput out) throws IOException {
    out.writeUTF(this.name);
    out.writeInt(this.age);
}

public void readFields(DataInput in) throws IOException {
    this.name = in.readUTF();
    this.age = in.readInt();

}
public int compareTo(StudentWritable o) {
return 0;
}
}
</div>

Combiner

一、Combiner的出现背景

1.1 回顾Map阶段五大步骤

我们认识了MapReduce的八大步凑，其中在Map阶段总共五个步骤，如下图所示：

map section

其中，step1.5是一个可选步骤，它就是我们今天需要了解的 Map规约阶段。现在，我们再来看看前一篇博文《[计数器与自定义计数器]》中的第一张关于计数器的图：

image

我们可以发现，其中有两个计数器：Combine output records和Combine input records，他们的计数都是0，这是因为我们在代码中没有进行Map阶段的规约操作。

1.2 为什么需要进行Map规约操作

众所周知，Hadoop框架使用Mapper将数据处理成一个个的<key,value>键值对，在网络节点间对其进行整理(shuffle)，然后使用Reducer处理数据并进行最终输出。

image

在上述过程中，我们看到至少两个性能瓶颈：

（1）如果我们有10亿个数据，Mapper会生成10亿个键值对在网络间进行传输，但如果我们只是对数据求最大值，那么很明显的Mapper只需要输出它所知道的最大值即可。这样做不仅可以减轻网络压力，同样也可以大幅度提高程序效率。

总结：网络带宽严重被占降低程序效率；

（2）假设使用美国专利数据集中的国家一项来阐述数据倾斜这个定义，这样的数据远远不是一致性的或者说平衡分布的，由于大多数专利的国家都属于美国，这样不仅Mapper中的键值对、中间阶段(shuffle)的键值对等，大多数的键值对最终会聚集于一个单一的Reducer之上，压倒这个Reducer，从而大大降低程序的性能。

总结：单一节点承载过重降低程序性能；

那么，有木有一种方案能够解决这两个问题呢？

二、初步探索Combiner

2.1 Combiner的横空出世

在MapReduce编程模型中，在Mapper和Reducer之间有一个非常重要的组件，它解决了上述的性能瓶颈问题，它就是Combiner。

PS：

①与mapper和reducer不同的是，combiner没有默认的实现，需要显式的设置在conf中才有作用。

②并不是所有的job都适用combiner，只有操作满足结合律的才可设置combiner。combine操作类似于：opt(opt(1, 2, 3), opt(4, 5, 6))。如果opt为求和、求最大值的话，可以使用，但是如果是求中值的话，不适用。

每一个map都可能会产生大量的本地输出，Combiner的作用就是对map端的输出先做一次合并，以减少在map和reduce节点之间的数据传输量，以提高网络IO性能，是MapReduce的一种优化手段之一，其具体的作用如下所述。

（1）Combiner最基本是实现本地key的聚合，对map输出的key排序，value进行迭代。如下所示：

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

（2）Combiner还有本地reduce功能（其本质上就是一个reduce），例如Hadoop自带的wordcount的例子和找出value的最大值的程序，combiner和reduce完全一致，如下所示：

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

PS：现在想想，如果在wordcount中不用combiner，那么所有的结果都是reduce完成，效率会相对低下。使用combiner之后，先完成的map会在本地聚合，提升速度。对于hadoop自带的wordcount的例子，value就是一个叠加的数字，所以map一结束就可以进行reduce的value叠加，而不必要等到所有的map结束再去进行reduce的value叠加。

2.2 融合Combiner的MapReduce

image

前面文章中的代码都忽略了一个可以优化MapReduce作业所使用带宽的步骤—Combiner，它在Mapper之后Reducer之前运行。Combiner是可选的，如果这个过程适合于你的作业，Combiner实例会在每一个运行map任务的节点上运行。Combiner会接收特定节点上的Mapper实例的输出作为输入，接着Combiner的输出会被发送到Reducer那里，而不是发送Mapper的输出。Combiner是一个“迷你reduce”过程，它只处理单台机器生成的数据。

2.3 使用MyReducer作为Combiner

在前面文章中的WordCount代码中加入以下一句简单的代码，即可加入Combiner方法：

// 设置Map规约Combiner
job.setCombinerClass(MyReducer.class)

还是以下面的文件内容为例，看看这次计数器会发生怎样的改变？

（1）上传的测试文件的内容

hello edison
hello kevin

（2）调试后的计数器日志信息

image

可以看到，原本都为0的Combine input records和Combine output records发生了改变。我们可以清楚地看到map的输出和combine的输入统计是一致的，而combine的输出与reduce的输入统计是一样的。由此可以看出规约操作成功，而且执行在map的最后，reduce之前。

三、自己定义Combiner

为了能够更加清晰的理解Combiner的工作原理，我们自定义一个Combiners类，不再使用MyReduce做为Combiners的类，具体的代码下面一一道来。

3.1 改写Mapper类的map方法

public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> { protected void map(LongWritable key, Text value,
Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws java.io.IOException, InterruptedException {
String line = value.toString();
String[] spilted = line.split(" "); for (String word : spilted) {
context.write(new Text(word), new LongWritable(1L)); // 为了显示效果而输出Mapper的输出键值对信息
System.out.println("Mapper输出<" + word + "," + 1 + ">");
}
};
}

3.2 改写Reducer类的reduce方法

public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> { protected void reduce(Text key,
java.lang.Iterable<LongWritable> values,
Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws java.io.IOException, InterruptedException { // 显示次数表示redcue函数被调用了多少次，表示k2有多少个分组
System.out.println("Reducer输入分组<" + key.toString() + ",N(N>=1)>"); long count = 0L; for (LongWritable value : values) {
count += value.get(); // 显示次数表示输入的k2,v2的键值对数量
System.out.println("Reducer输入键值对<" + key.toString() + ","
+ value.get() + ">");
}
context.write(key, new LongWritable(count));
};
}

3.3 添加MyCombiner类并重写reduce方法

public static class MyCombiner extends Reducer<Text, LongWritable, Text, LongWritable> { protected void reduce(
Text key,
java.lang.Iterable<LongWritable> values,
org.apache.hadoop.mapreduce.Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws java.io.IOException, InterruptedException { // 显示次数表示规约函数被调用了多少次，表示k2有多少个分组
System.out.println("Combiner输入分组<" + key.toString() + ",N(N>=1)>"); long count = 0L; for (LongWritable value : values) {
count += value.get(); // 显示次数表示输入的k2,v2的键值对数量
System.out.println("Combiner输入键值对<" + key.toString() + ","
+ value.get() + ">");
}
context.write(key, new LongWritable(count)); // 显示次数表示输出的k2,v2的键值对数量
System.out.println("Combiner输出键值对<" + key.toString() + "," + count + ">");
};
}

3.4 添加设置Combiner的代码

// 设置Map规约Combiner
job.setCombinerClass(MyCombiner.class);

3.5 调试运行的控制台输出信息

（1）Mapper

Mapper输出<hello,1> Mapper输出<edison,1> Mapper输出<hello,1> Mapper输出<kevin,1>

（2）Combiner

Combiner输入分组<edison,N(N>=1)> Combiner输入键值对<edison,1> Combiner输出键值对<edison,1> Combiner输入分组<hello,N(N>=1)> Combiner输入键值对<hello,1> Combiner输入键值对<hello,1> Combiner输出键值对<hello,2> Combiner输入分组<kevin,N(N>=1)> Combiner输入键值对<kevin,1> Combiner输出键值对<kevin,1></pre>

这里可以看出，在Combiner中进行了一次本地的Reduce操作，从而简化了远程Reduce节点的归并压力。

（3）Reducer

Reducer输入分组<edison,N(N>=1)> Reducer输入键值对<edison,1> Reducer输入分组<hello,N(N>=1)> Reducer输入键值对<hello,2> Reducer输入分组<kevin,N(N>=1)> Reducer输入键值对<kevin,1>

这里可以看出，在对hello的归并上，只进行了一次操作就完成了。

那么，如果我们再来看看不添加Combiner时的控制台输出信息：

（1）Mapper
Mapper输出<hello,1> Mapper输出<edison,1> Mapper输出<hello,1> Mapper输出<kevin,1>

（2）Reducer

Reducer输入分组<edison,N(N>=1)> Reducer输入键值对<edison,1> Reducer输入分组<hello,N(N>=1)> Reducer输入键值对<hello,1> Reducer输入键值对<hello,1> Reducer输入分组<kevin,N(N>=1)> Reducer输入键值对<kevin,1>

可以看出，没有采用Combiner时hello都是由Reducer节点来进行统一的归并，也就是这里为何会有两次hello的输入键值对了。

总结：从控制台的输出信息我们可以发现，其实combine只是把两个相同的hello进行规约，由此输入给reduce的就变成了<hello,2>。在实际的Hadoop集群操作中，我们是由多台主机一起进行MapReduce的，如果加入规约操作，每一台主机会在reduce之前进行一次对本机数据的规约，然后在通过集群进行reduce操作，这样就会大大节省reduce的时间，从而加快MapReduce的处理速度。
</div>

Partition分区

<div class='mdContent'>
旧版 API 的 Partitioner 解析

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

image

MapReduce 提供了两个Partitioner 实现：HashPartitioner和TotalOrderPartitioner。其中 HashPartitioner 是默认实现，它实现了一种基于哈希值的分片方法，代码如下：

<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; word-wrap: break-word; font-family: Consolas, "Courier New", 宋体, Courier, mono, serif; font-size: 12px !important; line-height: 1;">public int getPartition(K2 key, V2 value, int numReduceTasks) {
return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
}</pre>

TotalOrderPartitioner 提供了一种基于区间的分片方法，通常用在数据全排序中。在MapReduce 环境中，容易想到的全排序方案是归并排序，即在 Map 阶段，每个 Map Task进行局部排序；在 Reduce 阶段，启动一个 Reduce Task 进行全局排序。由于作业只能有一个 Reduce Task，因而 Reduce 阶段会成为作业的瓶颈。为了提高全局排序的性能和扩展性，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， bcd， efg， hii， mnk， rrr
如果 Reduce Task 个数为 4，则采样数据的四等分点为 abd、 bcd、 mnk，将这 3 个字符串作为分割点。
步骤2：Map 阶段。本阶段涉及两个组件，分别是 Mapper 和 Partitioner。其中，Mapper 可采用 IdentityMapper，直接将输入数据输出，但 Partitioner 必须选用TotalOrderPartitioner，它将步骤 1 中获取的分割点保存到 trie 树中以便快速定位任意一个记录所在的区间，这样，每个 Map Task 产生 R（Reduce Task 个数）个区间，且区间之间有序。TotalOrderPartitioner 通过 trie 树查找每条记录所对应的 Reduce Task 编号。如图所示，我们将分割点保存在深度为 2 的 trie 树中，假设输入数据中有两个字符串“ abg”和“ mnz”，则字符串“ abg” 对应 partition1，即第 2 个 Reduce Task，字符串“ mnz” 对应partition3，即第 4 个 Reduce Task。

image

步骤 3：Reduce 阶段。每个 Reducer 对分配到的区间数据进行局部排序，最终得到全排序数据。从以上步骤可以看出，基于 TotalOrderPartitioner 全排序的效率跟 key 分布规律和采样算法有直接关系；key 值分布越均匀且采样越具有代表性，则 Reduce Task 负载越均衡，全排序效率越高。TotalOrderPartitioner 有两个典型的应用实例： TeraSort 和 HBase 批量数据导入。其中，TeraSort 是 Hadoop 自带的一个应用程序实例。它曾在 TB 级数据排序基准评估中赢得第一名，而 TotalOrderPartitioner正是从该实例中提炼出来的。HBase 是一个构建在 Hadoop之上的 NoSQL 数据仓库。它以 Region为单位划分数据，Region 内部数据有序（按 key 排序），Region 之间也有序。很明显，一个 MapReduce 全排序作业的 R 个输出文件正好可对应 HBase 的 R 个 Region。

新版 API 的 Partitioner 解析

新版 API 中的Partitioner类图如图所示。它不再实现JobConfigurable 接口。当用户需要让 Partitioner通过某个JobConf 对象初始化时，可自行实现Configurable 接口，如：

image

Partition所处的位置

image

Partition主要作用就是将map的结果发送到相应的reduce。这就对partition有两个要求：

1）均衡负载，尽量的将工作均匀的分配给不同的reduce。

2）效率，分配速度一定要快。

Mapreduce提供的Partitioner

image

patition类结构

1. Partitioner<k,v>是partitioner的基类，如果需要定制partitioner也需要继承该类。源代码如下：

<p><code>Partitioner</code> controls the partitioning of the keys of the
intermediate map-outputs. The key (or a subset of the key) is used to derive
the partition, typically by a hash function. The total number of partitions
is the same as the number of reduce tasks for the job. Hence this controls
which of the <code>m</code> reduce tasks the intermediate key (and hence the
record) is sent for reduction.</p>
@see Reducer
@deprecated Use {@link org.apache.hadoop.mapreduce.Partitioner} instead. / @Deprecated public interface Partitioner<K2, V2> extends JobConfigurable { /* * Get the paritition number for a given key (hence record) given the total
- number of partitions i.e. number of reduce-tasks for the job.
- <p>Typically a hash function on a all or a subset of the key.</p>
- @param key the key to be paritioned.
- @param value the entry value.
- @param numPartitions the total number of partitions.
- @return the partition number for the <code>key</code>. */
  int getPartition(K2 key, V2 value, int numPartitions);
  }</pre>

2. HashPartitioner<k,v>是mapreduce的默认partitioner。源代码如下：

<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; word-wrap: break-word; font-family: Consolas, "Courier New", 宋体, Courier, mono, serif; font-size: 12px !important; line-height: 1;">package org.apache.hadoop.mapreduce.lib.partition; import org.apache.hadoop.mapreduce.Partitioner; /** Partition keys by their {@link Object#hashCode()}. /
public class HashPartitioner<K, V> extends Partitioner<K, V> { /* Use {@link Object#hashCode()} to partition. */
public int getPartition(K key, V value, int numReduceTasks) { return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
}
}</pre>

3. BinaryPatitioner继承于Partitioner<BinaryComparable ,V>，是Partitioner<k,v>的偏特化子类。该类提供leftOffset和rightOffset，在计算which reducer时仅对键值K的[rightOffset，leftOffset]这个区间取hash。

4. KeyFieldBasedPartitioner<k2, v2="">也是基于hash的个partitioner。和BinaryPatitioner不同，它提供了多个区间用于计算hash。当区间数为0时KeyFieldBasedPartitioner退化成HashPartitioner。源代码如下：

<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; word-wrap: break-word; font-family: Consolas, "Courier New", 宋体, Courier, mono, serif; font-size: 12px !important; line-height: 1;">package org.apache.hadoop.mapred.lib; import java.io.UnsupportedEncodingException; import java.util.List; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.Partitioner; import org.apache.hadoop.mapred.lib.KeyFieldHelper.KeyDescription; /** * Defines a way to partition keys based on certain key fields (also see

{@link KeyFieldBasedComparator}.
The key specification supported is of the form -k pos1[,pos2], where,
pos is of the form f[.c][opts], where f is the number
of the key field to use, and c is the number of the first character from
the beginning of the field. Fields and character posns are numbered
starting with 1; a character position of zero in pos2 indicates the
field's last character. If '.c' is omitted from pos1, it defaults to 1
(the beginning of the field); if omitted from pos2, it defaults to 0
(the end of the field).
*/
public class KeyFieldBasedPartitioner<K2, V2> implements Partitioner<K2, V2> { private static final Log LOG = LogFactory.getLog(KeyFieldBasedPartitioner.class.getName()); private int numOfPartitionFields; private KeyFieldHelper keyFieldHelper = new KeyFieldHelper(); public void configure(JobConf job) {
String keyFieldSeparator = job.get("map.output.key.field.separator", "\t");
keyFieldHelper.setKeyFieldSeparator(keyFieldSeparator); if (job.get("num.key.fields.for.partition") != null) {
LOG.warn("Using deprecated num.key.fields.for.partition. " +
"Use mapred.text.key.partitioner.options instead"); this.numOfPartitionFields = job.getInt("num.key.fields.for.partition",0);
keyFieldHelper.setKeyFieldSpec(1,numOfPartitionFields);
} else {
String option = job.getKeyFieldPartitionerOption();
keyFieldHelper.parseOption(option);
}
} public int getPartition(K2 key, V2 value, int numReduceTasks) { byte[] keyBytes;

List <KeyDescription> allKeySpecs = keyFieldHelper.keySpecs(); if (allKeySpecs.size() == 0) { return getPartition(key.toString().hashCode(), numReduceTasks);
} try {
keyBytes = key.toString().getBytes("UTF-8");
} catch (UnsupportedEncodingException e) { throw new RuntimeException("The current system does not " +
"support UTF-8 encoding!", e);
} // return 0 if the key is empty
if (keyBytes.length == 0) { return 0;
} int []lengthIndicesFirst = keyFieldHelper.getWordLengths(keyBytes, 0,
keyBytes.length); int currentHash = 0; for (KeyDescription keySpec : allKeySpecs) { int startChar = keyFieldHelper.getStartOffset(keyBytes, 0, keyBytes.length,
lengthIndicesFirst, keySpec); // no key found! continue
if (startChar < 0) { continue;
} int endChar = keyFieldHelper.getEndOffset(keyBytes, 0, keyBytes.length,
lengthIndicesFirst, keySpec);
currentHash = hashCode(keyBytes, startChar, endChar,
currentHash);
} return getPartition(currentHash, numReduceTasks);
} protected int hashCode(byte[] b, int start, int end, int currentHash) { for (int i = start; i <= end; i++) {
currentHash = 31*currentHash + b[i];
} return currentHash;
} protected int getPartition(int hash, int numReduceTasks) { return (hash & Integer.MAX_VALUE) % numReduceTasks;
}
}</pre>

5. TotalOrderPartitioner这个类可以实现输出的全排序。不同于以上3个partitioner，这个类并不是基于hash的。下面详细的介绍TotalOrderPartitioner

TotalOrderPartitioner 类

每一个reducer的输出在默认的情况下都是有顺序的，但是reducer之间在输入是无序的情况下也是无序的。如果要实现输出是全排序的那就会用到TotalOrderPartitioner。

要使用TotalOrderPartitioner，得给TotalOrderPartitioner提供一个partition file。这个文件要求Key（这些key就是所谓的划分）的数量和当前reducer的数量-1相同并且是从小到大排列。对于为什么要用到这样一个文件，以及这个文件的具体细节待会还会提到。

TotalOrderPartitioner对不同Key的数据类型提供了两种方案：

1）对于非BinaryComparable 类型的Key，TotalOrderPartitioner采用二分发查找当前的K所在的index。

例如：reducer的数量为5，partition file 提供的4个划分为【2，4，6，8】。如果当前的一个key/value 是<4,”good”>，利用二分法查找到index=1，index+1=2那么这个key/value 将会发送到第二个reducer。如果一个key/value为<4.5, “good”>。那么二分法查找将返回-3，同样对-3加1然后取反就是这个key/value将要去的reducer。

对于一些数值型的数据来说，利用二分法查找复杂度是O(log(reducer count))，速度比较快。

2）对于BinaryComparable类型的Key（也可以直接理解为字符串）。字符串按照字典顺序也是可以进行排序的。

这样的话也可以给定一些划分，让不同的字符串key分配到不同的reducer里。这里的处理和数值类型的比较相近。

例如：reducer的数量为5，partition file 提供了4个划分为【“abc”, “bce”, “eaa”, ”fhc”】那么“ab”这个字符串将会被分配到第一个reducer里，因为它小于第一个划分“abc”。

但是不同于数值型的数据，字符串的查找和比较不能按照数值型数据的比较方法。mapreducer采用的Tire tree（关于Tire tree可以参考《字典树(Trie Tree)》）的字符串查找方法。查找的时间复杂度o(m)，m为树的深度，空间复杂度o(255^m-1)。是一个典型的空间换时间的案例。

Tire tree的构建

假设树的最大深度为3，划分为【aaad ，aaaf， aaaeh，abbx】

image

Mapreduce里的Tire tree主要有两种节点组成：

1） Innertirenode
Innertirenode在mapreduce中是包含了255个字符的一个比较长的串。上图中的例子只包含了26个英文字母。
2）叶子节点{unslipttirenode, singesplittirenode, leaftirenode}
Unslipttirenode 是不包含划分的叶子节点。
Singlesplittirenode 是只包含了一个划分点的叶子节点。
Leafnode是包含了多个划分点的叶子节点。（这种情况比较少见，达到树的最大深度才出现这种情况。在实际操作过程中比较少见）

Tire tree的搜索过程

接上面的例子：
1）假如当前 key value pair <aad, 10="">这时会找到图中的leafnode，在leafnode内部使用二分法继续查找找到返回 aad在划分数组中的索引。找不到会返回一个和它最接近的划分的索引。
2）假如找到singlenode，如果和singlenode的划分相同或小返回他的索引，比singlenode的划分大则返回索引+1。
3）假如找到nosplitnode则返回前面的索引。如<zaa, 20="">将会返回abbx的在划分数组中的索引。

TotalOrderPartitioner的疑问

上面介绍了partitioner有两个要求，一个是速度，另外一个是均衡负载。使用tire tree提高了搜素的速度，但是我们怎么才能找到这样的partition file 呢？让所有的划分刚好就能实现均衡负载。

InputSampler
输入采样类，可以对输入目录下的数据进行采样。提供了3种采样方法。

image

采样类结构图

采样方式对比表:

类名称

采样方式

构造方法

效率

特点

|
|

SplitSampler<K,V>

对前n个记录进行采样

采样总数，划分数

最高

| |
|

RandomSampler<K,V>

遍历所有数据，随机采样

采样频率，采样总数，划分数

最低

| |
|

IntervalSampler<K,V>

固定间隔采样

采样频率，划分数

中

对有序的数据十分适用

writePartitionFile这个方法很关键，这个方法就是根据采样类提供的样本，首先进行排序，然后选定（随机的方法）和reducer数目-1的样本写入到partition file。这样经过采样的数据生成的划分，在每个划分区间里的key/value就近似相同了，这样就能完成均衡负载的作用。

SplitSampler类的源代码如下：

Inexpensive way to sample random data. /
public static class SplitSampler<K,V> implements Sampler<K,V> { private final int numSamples; private final int maxSplitsSampled; /* * Create a SplitSampler sampling <em>all</em> splits.
- Takes the first numSamples / numSplits records from each split.
- @param numSamples Total number of samples to obtain from all selected
- ```
              splits. */
```

public SplitSampler(int numSamples) { this(numSamples, Integer.MAX_VALUE);
} /** * Create a new SplitSampler.
 * @param numSamples Total number of samples to obtain from all selected
 *                   splits.
 * @param maxSplitsSampled The maximum number of splits to examine. */
public SplitSampler(int numSamples, int maxSplitsSampled) { this.numSamples = numSamples; this.maxSplitsSampled = maxSplitsSampled;
} /** * From each split sampled, take the first numSamples / numSplits records. */ @SuppressWarnings("unchecked") // ArrayList::toArray doesn't preserve type
public K[] getSample(InputFormat<K,V> inf, JobConf job) throws IOException {
  InputSplit[] splits = inf.getSplits(job, job.getNumMapTasks());
  ArrayList<K> samples = new ArrayList<K>(numSamples); int splitsToSample = Math.min(maxSplitsSampled, splits.length); int splitStep = splits.length / splitsToSample; int samplesPerSplit = numSamples / splitsToSample; long records = 0; for (int i = 0; i < splitsToSample; ++i) {
    RecordReader<K,V> reader = inf.getRecordReader(splits[i * splitStep],
        job, Reporter.NULL);
    K key = reader.createKey();
    V value = reader.createValue(); while (reader.next(key, value)) {
      samples.add(key);
      key = reader.createKey(); ++records; if ((i+1) * samplesPerSplit <= records) { break;
      }
    }
    reader.close();
  } return (K[])samples.toArray();
}

}</pre>

RandomSampler类的源代码如下：

General-purpose sampler. Takes numSamples / maxSplitsSampled inputs from
each split. /
public static class RandomSampler<K,V> implements Sampler<K,V> { private double freq; private final int numSamples; private final int maxSplitsSampled; /* * Create a new RandomSampler sampling <em>all</em> splits.
- This will read every split at the client, which is very expensive.
- @param freq Probability with which a key will be chosen.
- @param numSamples Total number of samples to obtain from all selected
- ```
              splits. */
```

public RandomSampler(double freq, int numSamples) { this(freq, numSamples, Integer.MAX_VALUE);
} /** * Create a new RandomSampler.
 * @param freq Probability with which a key will be chosen.
 * @param numSamples Total number of samples to obtain from all selected
 *                   splits.
 * @param maxSplitsSampled The maximum number of splits to examine. */
public RandomSampler(double freq, int numSamples, int maxSplitsSampled) { this.freq = freq; this.numSamples = numSamples; this.maxSplitsSampled = maxSplitsSampled;
} /** * Randomize the split order, then take the specified number of keys from
 * each split sampled, where each key is selected with the specified
 * probability and possibly replaced by a subsequently selected key when
 * the quota of keys from that split is satisfied. */ @SuppressWarnings("unchecked") // ArrayList::toArray doesn't preserve type
public K[] getSample(InputFormat<K,V> inf, JobConf job) throws IOException {
  InputSplit[] splits = inf.getSplits(job, job.getNumMapTasks());
  ArrayList<K> samples = new ArrayList<K>(numSamples); int splitsToSample = Math.min(maxSplitsSampled, splits.length);

  Random r = new Random(); long seed = r.nextLong();
  r.setSeed(seed);
  LOG.debug("seed: " + seed); // shuffle splits
  for (int i = 0; i < splits.length; ++i) {
    InputSplit tmp = splits[i]; int j = r.nextInt(splits.length);
    splits[i] = splits[j];
    splits[j] = tmp;
  } // our target rate is in terms of the maximum number of sample splits, // but we accept the possibility of sampling additional splits to hit // the target sample keyset
  for (int i = 0; i < splitsToSample || (i < splits.length && samples.size() < numSamples); ++i) {
    RecordReader<K,V> reader = inf.getRecordReader(splits[i], job,
        Reporter.NULL);
    K key = reader.createKey();
    V value = reader.createValue(); while (reader.next(key, value)) { if (r.nextDouble() <= freq) { if (samples.size() < numSamples) {
          samples.add(key);
        } else { // When exceeding the maximum number of samples, replace a // random element with this one, then adjust the frequency // to reflect the possibility of existing elements being // pushed out
          int ind = r.nextInt(numSamples); if (ind != numSamples) {
            samples.set(ind, key);
          }
          freq *= (numSamples - 1) / (double) numSamples;
        }
        key = reader.createKey();
      }
    }
    reader.close();
  } return (K[])samples.toArray();
}

}</pre>

IntervalSampler类的源代码为：

Useful for sorted data. /
public static class IntervalSampler<K,V> implements Sampler<K,V> { private final double freq; private final int maxSplitsSampled; /* * Create a new IntervalSampler sampling <em>all</em> splits.
- @param freq The frequency with which records will be emitted. /
  public IntervalSampler(double freq) { this(freq, Integer.MAX_VALUE);
  } /* * Create a new IntervalSampler.
- @param freq The frequency with which records will be emitted.
- @param maxSplitsSampled The maximum number of splits to examine.
- @see #getSample /
  public IntervalSampler(double freq, int maxSplitsSampled) { this.freq = freq; this.maxSplitsSampled = maxSplitsSampled;
  } /* * For each split sampled, emit when the ratio of the number of records
- retained to the total record count is less than the specified
- frequency. */ @SuppressWarnings("unchecked") // ArrayList::toArray doesn't preserve type
  public K[] getSample(InputFormat<K,V> inf, JobConf job) throws IOException {
  InputSplit[] splits = inf.getSplits(job, job.getNumMapTasks());
  ArrayList<K> samples = new ArrayList<K>(); int splitsToSample = Math.min(maxSplitsSampled, splits.length); int splitStep = splits.length / splitsToSample; long records = 0; long kept = 0; for (int i = 0; i < splitsToSample; ++i) {
  RecordReader<K,V> reader = inf.getRecordReader(splits[i * splitStep],
  job, Reporter.NULL);
  K key = reader.createKey();
  V value = reader.createValue(); while (reader.next(key, value)) { ++records; if ((double) kept / records < freq) { ++kept;
  samples.add(key);
  key = reader.createKey();
  }
  }
  reader.close();
  } return (K[])samples.toArray();
  }
  }</pre>

InputSampler类完整源代码如下：

image

InputSampler

TotalOrderPartitioner实例

<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; word-wrap: break-word; font-family: Consolas, "Courier New", 宋体, Courier, mono, serif; font-size: 12px !important; line-height: 1;">public class SortByTemperatureUsingTotalOrderPartitioner extends Configured implements Tool
{
@Override public int run(String[] args) throws Exception
{
JobConf conf = JobBuilder.parseInputAndOutput(this, getConf(), args); if (conf == null) { return -1;
}
conf.setInputFormat(SequenceFileInputFormat.class);
conf.setOutputKeyClass(IntWritable.class);
conf.setOutputFormat(SequenceFileOutputFormat.class);
SequenceFileOutputFormat.setCompressOutput(conf, true);
SequenceFileOutputFormat
.setOutputCompressorClass(conf, GzipCodec.class);
SequenceFileOutputFormat.setOutputCompressionType(conf,
CompressionType.BLOCK);
conf.setPartitionerClass(TotalOrderPartitioner.class);
InputSampler.Sampler<IntWritable, Text> sampler = new InputSampler.RandomSampler<IntWritable, Text>( 0.1, 10000, 10);
Path input = FileInputFormat.getInputPaths(conf)[0];
input = input.makeQualified(input.getFileSystem(conf));
Path partitionFile = new Path(input, "_partitions");
TotalOrderPartitioner.setPartitionFile(conf, partitionFile);
InputSampler.writePartitionFile(conf, sampler); // Add to DistributedCache
URI partitionUri = new URI(partitionFile.toString() + "#_partitions");
DistributedCache.addCacheFile(partitionUri, conf);
DistributedCache.createSymlink(conf);
JobClient.runJob(conf); return 0;
} public static void main(String[] args) throws Exception { int exitCode = ToolRunner.run( new SortByTemperatureUsingTotalOrderPartitioner(), args);
System.exit(exitCode);
}
}</pre>

</div>

MapReduce工作机制和序列化

MapReduce执行流程

MapReduce中的序列化

hadoop序列化的作用: <div class="mdContent">

Combiner

一、Combiner的出现背景

1.1 回顾Map阶段五大步骤

1.2 为什么需要进行Map规约操作

二、初步探索Combiner

2.1 Combiner的横空出世

2.2 融合Combiner的MapReduce

2.3 使用MyReducer作为Combiner

三、自己定义Combiner

3.1 改写Mapper类的map方法

3.2 改写Reducer类的reduce方法

3.3 添加MyCombiner类并重写reduce方法

3.4 添加设置Combiner的代码

3.5 调试运行的控制台输出信息

Partition分区

TotalOrderPartitioner 类

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读