《权威指南》笔记八 MapReduce的特性

作者：JesseZhang （CNZQS|JesseZhang）

博客地址：http://www.cnzqs.com

 

要点：

1、计数器

1.1内置计数器

1.2用户定义的Java计数器

1.3用户定义的Streaming计数器

2、排序

2.1准备

2.2部分排序

2.3全排序

2.4辅助排序

3、连接

3.1map端连接

3.2reduce端连接

4、边数据分布

3.1利用JobConf来配置作业

3.2分布式缓存

5、MapReduce库类

 

============================================

1、 计数器

 

计数器主要用来收集系统信息和作业运行信息，用于知道作业成功、失败等情况，比日志更便利进行分析。

 

1.1内置计数器

Hadoop内置的计数器，记录作业执行情况和记录情况。包括MapReduce框架、文件系统、作业计数三大类。

 

计数器由关联任务维护，定期传递给tasktracker，再由tasktracker传给jobtracker。

计数器可以被全局聚集。内置的作业计数器实际上由jobtracker维护，不必在整个网络中传递。

当一个作业执行成功后，计数器的值才是完整可靠的。

1.2用户定义的Java计数器

MapReduce框架将跨所有map和reduce聚集这些计数器，并在作业结束时产生一个最终结果。

多个计数器由一个Java枚举类型来定义，以便对计数器分组。

通过reporter输出和写入。在0.20.2之上版本中使用context.getCounter(groupName, counterName) 来获取计数器配置并设置。

动态计数器

动态计数器，incrCounter(String group,String counter,long amount)

组名称，计数器名称，计数值

 

尽量为计数器提供易读的名称，以Java枚举类型为名创建一个属性文件，用下划线分割嵌套类型。属性文件与包含该枚举类型的顶级类放在同一目录。

CounterGroupName = Air Temperature Records          //只有一个该属性，就是组的显示名称

MISSING.name = Missing                 //字段名称.name 格式

MALFORMED.name = malformed

 

可以采用本地化机制

 

获取计数器：

两种获取计数器的方法：

1、  hadoop job –counter 指令

2、  通过java API，在作业运行完成后，计数器稳定后获取。 使用job.getCounters()得到Counters

 

可能的问题：无法获取到有效的RunningJob对象，原因可能是：

1、  错误的指定了jobid

2、  内存中仅保留最新的100个作业（可以在mapred.jobtracker.completeuserjobs.maximun控制）

 

1.3用户定义的Streaming计数器

 

Streaming MapReduce程序通过向标准错误流发送一行特殊格式的信息来增加计数器的值，格式如下：

Reporter:counter:group,counter,amount

 

Python用法：

Sys.stderr.write(“reporter:counter:Temperature,Missing,1\n”)

 

状态信息：

Reporter:status:mesage

 

2、 排序

排序是MapReduce的核心技术。

         2.1准备

每个map创建并输出一个块压缩的顺序文件。

 

         2.2部分排序

控制排列顺序

1、  若属性mapred.output.key.comparator.class设置，则使用该类实例

2、  否则，键必须是WritableComparable的子类。

 

应用：基于分区的MapFile查找技术。

 

         2.3全排序

只有一个分区可以实现，但无法利用并行的优势。

 

替代方案：

1、  创建一系列排好序的文件

2、  串联这些文件，

3、  生成一个全局排序的文件

思路：使用一个partitioner来描述全局排序的输出。

关键点在于如何划分各个分区。尽量使各分区的记录数相等。

1、  可以写一个MapReduce程序来计算各个范围数据的分布，用来做评估。但性能较低

2、  可以通过对键空间进行采样，获取键的近似分布。

 

Hadoop内置了采样器，不需要自己编写。

Input Sampler类实现了Sampler接口，getSampler方法，返回一系列的样本键。

通常不直接由客户端调用，而是由InputSampler类的静态方法writePartitionFile()调用，目的是创建一个顺序文件来存储定义分区的键。

顺序文件供TotalOrderPartitioner使用，为排序作业创建分区。

 

2.4辅助排序

主要解决对键所对应的值的排序。

1、  定义包括自然键和自然值的组合键

2、  键的comparator根据组合键对记录进行排序，即同时利用自然键和自然值进行排序

3、  针对组合键的comparator和分组comparator在进行分区和分组时均只考虑自然键

 

3、 连接

 

连接操作的具体实现技术取决于数据集的规模和分区方式。

1、  如果一个数据集很大，一个数据集很小，则将集合分发到集群中每一个节点，从而实现连接

2、  对两个数据集都很大，则需要根据数据的组织方式，来决定是在map端连接还是在reduce端连接。

 

         3.1map端连接

在两个大规模输入数据集之间的map端连接会在数据到达map函数之前就执行连接操作。

 

         3.2reduce端连接

Reduce端连接并不要求输入数据集符合特定结构，因而reduce端连接比ap端连接更为常用。但由于两个数据集均需经过MapReduce的shuffle过程，所以reduce端连接的效率往往更低一些。

基本思路：mapper为各个记录标记源，并且使用连接健作为map输出键，使键相同的记录放在同一个reducer中。

主要技术有：

多输出：MultipleInputs

辅助排序：

 

4、 边数据分布

“边数据”（side data）是作业所需的额外的只读数据，以辅助处理主数据集。

 

         3.1利用JobConf来配置作业

如果只需向任务传递少量元数据则非常有用。如果想获取任务的值，只需覆盖Mapper或Reducer类的configure（）方法，并调用传入JobConf对象的getter方法即可。

 

         3.2分布式缓存

Hadoop的分布式缓存拷贝机制，它能够在任务运行过程中及时将文件和存档复制到任务节点以供使用。为了节约网络带宽，在每一个作业中，各个文件通常只需复制到一个节点一次。

缓存的容量是有限的，默认10G，可以在local.cache.size进行配置

 

工作机制：

1、  用户启动一个作业，hadoop将由-files 和 –archieves选项所指定的文件复制到jobtracker的文件系统中。

2、  在任务运行之前，tasktracker将文件从jobtracker的文件系统中复制到本地磁盘—缓存—使任务能够访问文件。

 

Tasktracker为缓存中的文件维护一个计数器统计使用情况，如果计数器为0，则清除。

 

分布式缓存的API：DistributedCache API

通过GenericOptionsParser间接使用分布式缓存。

方法：addCachedFile() 和 addCacheArchive() 方法告诉分布式缓存在HDFS中的位置。

 

获取缓存中的有效文件的列表：

方法：getLocalCachedFiles() 和 getLocalCacheArchives() 返回一个指向本地文件路径对象数组。

 

5、 MapReduce库类

1、  ChainMapper、ChainReducer

2、  FieldSelectionMapReduce

3、  IntSumRecuder LongSumReducer

4、  InverseMapper

5、  TokenCounterMapper

6、  RegexMapper

 

-----------------------------------------

转载请注明来源：CNZQS|JesseZhang的个人博客-《《权威指南》笔记八 MapReduce的特性》

本文链接地址：http://www.cnzqs.com/?p=105