读书笔记2：Hadoop组件-（2）剖析MapReduce程序

 

3，剖析 MapReduce 程序

<1>hadoop的数据类型

实现Writable 接口额类可以是值，而实现 WritableComparable 接口的类既可以是键也可以是值。

以下这些是常用的数据类型，均用于实现WritableComparable 借口：

BooleanWritable

ByteWritable

DoubleWritable

FloatWritable

IntWritable

LongWritable

Text 使用UTF8 格式的文本封装

NullWritable 无键值时的占位符

也可以自定义数据类型，但要实现Writable 接口或者 WritableComparable 接口。

<2>Mapper

一个类要作为mapper ，需要继承 MapReduceBase 基类和实现 Mapper 借口。

mapper和 reducer 的基类均为 MapReduceBase ，它包含类的构造和解构方法：

void configure(JobConf job);该函数提取 XML 配置文件或者应用程序主类中的参数，在数据处理之前调用该函。

void close();作为 map 任务结束前的最后一个操作，该函数完成所有的结尾工作，如关闭数据库、打开文件等。

Mapper借口负责数据处理阶段，采用的形式是 Mapper<K1,V1,K2,V2>java 泛型，键类和值类分别实现 WritableComparable 和 Writable 借口。

Mapper只有一个方法 map 用于处理一个单独的键值对。

void map(K1 key,V1 value,OutputCollector<K2,V2> output,Reporter reporter)

该函数处理一个给定的键值对(K1,V1), 生成一个键值对 (K2,V2) 的列表。

OutputCollector接收这个映射过程的输入。

Reporter 可提供对 Mapper 相关附加信息的记录，形成任务进度。

 

Hadoop预定义的 Mapper 的实现 :

IdentityMapper<K,V>	实现Mapper<K,V,K,V> 就输入直接映射到输出
InverseMapper<K,V>	实现Mapper<K,V,K,V> 发转键值对
RegexMapper<K>	实现Mapper<K,Text,Text,LongWritable>, 为每个常规表达式的匹配项生成一个 (mathc,1) 对
TokenCountMapper<K>	实现Mapper<K,Text,Text,LongWritable>, 当输入的值为分词时，生成一个 (token,1) 对

<3>Reducer

reducer的实现也必须继承 MapReduceBase 类和实现 Reducer 接口。

Reducer接口只有一个方法，及时 reduce:

Void reduce (K2,key,Iterator<V2> values,OutputCollector<K3,V3> output,Reporter r)

Reducer任务接收来自各个 mapper 的输出时，按照键对输入数据进行排序，并将相同的键的值归并，然后调用 reduce 函数，并通过迭代处理那些与指定键相关联的值，生成一个列表 <K3,V3>

OutputCollector接收 reduce 阶段的输出，并写入输出文件。

Reporter 可提供对 Mapper 相关附加信息的记录，形成任务进度。

 

Hadoop预定义的 Reducer 的实现 :

IdentityReducer<K,V>	实现Reducer<K,V,K,V> ，将输入直接映射到输出
LongSumReducer<K>	实现<K,LongWritable,K,LongWritable>, 计算与给定键相对应的所有值的和

<4>Partitioner：重定向 Mapper 输出

其实在mapper 和 reducer 之间有一个非常重要的环节，就是将 mapper 的结果输出给不同的 reducer ，这就是 Partitioner 的工作：（此工作常被称为： 洗牌 ）

一般reducer 是多个，那么 mapper 应该将键值的输出给谁呢？ Hadoop 默认的机制是对键进行散列来确定 reducer ， Hadoop 通过 HashPartitioner 类强制执行此策略。但是有的时候 HashPartitioner 会使得程序出错，即他的分发策略不符合实际的要求，那么此时就需要我们定制自己的 Partitioner ： ( 例如针对自定义的数据类型 Edge)

public   class  EdgePartitioner  implements  Partitioner<Edge,Writable>

{

@override

public   int  getPartition(Edge key,Writable value, int  numPartitions)

{

return  key.getDepartureNode().hashCode() % numPartitions;

}

@override

public   void  configure(JobConf conf)

{

}

}

自己定制的Partitioner 只需要实现 getPartition 方法和 configure 方法即可。前者将 Hadoop对作业的配置应用在 Partitioner上，而后者返回一个介于 0 和 reducer 任务数之间的整数，指向键值对要发送到的 reducer 。

Partitioner的形象的表达：（如下图）

<5>Combiner :本地 reduce

在分发mapper 之前先做一下“本地 reduce ”，也被成为合并，后面详细讲述。

<6>预定义 Mapper 和 Reducer

使用预定义Mapper 和 Reducer 改写前面的统计单词数量的程序： WordCount2.java

public   class  WordCount2 {

public   static   void  main(String args[])

{

JobClient client =  new  JobClient();

JobConf conf =   new  JobConf(WordCount2. class );

FileInputFormat.addInputPath(conf, new  Path(args[0]));

FileOutputFormat.setOutputPath(conf, new  Path(args[1]));

conf.setOutputKeyClass(Text. class );

conf.setOutputValueClass(LongWritable. class );

conf.setMapperClass(TokenCountMapper. class ); //hadoop自己的TokenCountMapper

conf.setCombinerClass(LongSumReducer. class ); //hadoop自己的LongSumReducer

conf.setReducerClass(LongSumReducer. class );

client.setConf(conf);

try

{

JobClient.runJob(conf);

} catch (Exception e)

{

e.printStackTrace();

}

}

}