Hadoop MapReduce高级编程

标签: hadoop mapreduce 编程 | 发表时间:2014-02-21 11:18 | 作者:haiyang_1987
出处:http://www.iteye.com
转载:
http://www.cnblogs.com/end/archive/2013/01/18/2866824.html



第一部分:重要的组件
Combiner
•什么是Combiner
•combine函数把一个map函数产生的<key,value>对(多个key, value)合并成一个新的<key2,value2>. 将新的<key2,value2>作为输入到reduce函数中,其格式与reduce函数相同。
•这样可以有效的较少中间结果,减少网络传输负荷。

•什么情况下可以使用Combiner
•可以对记录进行汇总统计的场景,如求和。
•求平均数的场景就不可以使用了
Combiner执行时机
•运行combiner函数的时机有可能会是merge完成之前,或者之后,这个时机可以由一个参数控制,即 min.num.spill.for.combine(default 3)
•当job中设定了combiner,并且spill数最少有3个的时候,那么combiner函数就会在merge产生结果文件之前运行
•通过这样的方式,就可以在spill非常多需要merge,并且很多数据需要做conbine的时候,减少写入到磁盘文件的数据数量,同样是为了减少对磁盘的读写频率,有可能达到优化作业的目的。
•Combiner也有可能不执行, Combiner会考虑当时集群的负载情况。
Combiner如何使用
•代码示例
•继承Reducer类
public static class Combiner extends MapReduceBase implements
           Reducer<Text, Text, Text, Text> {
       public void reduce(Text key, Iterator<Text> values,
               OutputCollector<Text, Text> output, Reporter reporter)
               throws IOException {
                 }
    }

•配置作业时加入conf.setCombinerClass(Combiner.class)

Partitioner
•什么是Partitioner
•Mapreduce 通过Partitioner 对Key 进行分区,进而把数据按我们自己的需求来分发。
•什么情况下使用Partitioner
•如果你需要key按照自己意愿分发,那么你需要这样的组件。
•例如:数据文件内包含省份,而输出要求每个省份输出一个文件。
•框架默认的HashPartitioner
•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; 
  }
}
Partitioner如何使用
•实现Partitioner接口覆盖getPartition()方法
•配置作业时加入conf.setPartitionerClass(MyPartitioner.class);
•Partitioner示例
        public static class MyPartitioner implements Partitioner<Text, Text> {
          
         @Override
            public int getPartition(Text key, Text value, int numPartitions) {
             }

}
Partitioner需求示例
•需求描述
•数据文件中含有省份
•需要相同的省份送到相同的Reduce里
•从而产生不同的文件
•数据样例
•1 liaoning
•1 代表该省份有多少个直辖市
•步骤
•实现Partitioner,覆盖getPartition
•根据省份字段进行切分


RecordReader
•什么是RecordReader
•用于在分块中读取<Key,Value>对,也就是说每一次我们读取一条记录都会调用该类。
•主要是处理经过InputFormat分片完的数据
•什么时候使用RecordReader
•需要对输入的数据按自己的需求处理
•如:要求输入的key不是文件的偏移量而是文件的路径或者名字
•系统默认为LineRecordReader
•按照每行的偏移量做为map输出时的key值,每行的内容作为map的value值,默认的分隔符是回车和换行。

RecordReader需求示例
•需求
•更改map对应的输入的<key,value>值,key对应的文件的路径(或者是文件名),value对应的是文件的内容(content)。
•步骤
•重写InputFormat不对文件切分
•重写RecordReader
•在配置作业时使用自定义的组件进行数据处理


第二部分:Join
案例分析
•输入为2个文件,文件一内容如下
•空格分割:用户名 手机号 年龄
•内容样例
•Tom 1314567890 14
•文件二内容
•空格分割:手机号 地市
•内容样例
•13124567890 hubei
•需要统计出的汇总信息为 用户名 手机号 年龄 地市
Map端Join
•设计思路
•使用DistributedCache.addCacheFile()将地市的文件加入到所有Map的缓存里
•在Map函数里读取该文件,进行Join
•  将结果输出到reduce
•需要注意的是
•DistributedCache需要在生成Job作业前使用


Reduce端Join
•设计思路
•Map端读取所有文件,并在输出的内容里加上标识代表数据时从哪个文件里来的
•在reduce对按照标识对数据进行保存
•然后根据Key的Join来求出结果直接输出

第三部分:排序

普通排序
•Mapreduce本身自带排序功能
•Text对象是不适合排序的,如果内容为整型不会安照编码顺序去排序
•一般情况下我们可以考虑以IntWritable做为Key,同时将Reduce设置成0 ,进行排序

部分排序
•即输出的每个文件都是排过序的
•如果我们不需要全局排序,那么这是个不错的选择。

全局排序
•产生背景
•Hadoop平台没有提供全局数据排序,而在大规模数据处理中进行数据的全局排序是非常普遍的需求。
•使用hadoop进行大量的数据排序排序最直观的方法是把文件所有内容给map之后,map不做任何处理,直接输出给一个reduce,利用hadoop的自己的shuffle机制,对所有数据进行排序,而后由reduce直接输出。
•快速排序基本步骤就是需要现在所有数据中选取一个作为支点。然后将大于这个支点的放在一边,小于这个支点的放在另一边。

设想如果我们有 N 个支点(这里可以称为标尺),就可以把所有的数据分成 N+1 个 part ,将这 N+1 个 part 丢给 reduce ,由 hadoop 自动排序,最后输出 N+1 个内部有序的文件,再把这 N+1 个文件首尾相连合并成一个文件,收工 。
由此我们可以归纳出这样一个用 hadoop 对大量数据排序的步骤:
1 )   对待排序数据进行抽样;
2 )   对抽样数据进行排序,产生标尺;
3 )   Map 对输入的每条数据计算其处于哪两个标尺之间;将数据发给对应区间 ID 的 reduce
4 )   Reduce 将获得数据直接输出。
•Hadoop 提供了Sampler接口可以返回一组样本,该接口为Hadoop的采样器。
           public interface Sampler<K, V> {
                        K[] getSample(InputFormat<K, V> inf, Job job)
                         throws IOException, InterruptedException;
            }
•Hadoop提供了一个TotalOrderPartitioner,可以使我们来实现全局排序。
二次排序
•产生背景
•MapReduce默认会对key进行排序
•将输出到Reduce的values也进行预先的排序
•实现方式
•重写Partitioner,完成key分区,进行第一次排序
•实现WritableComparator,完成自己的排序逻辑,完成key的第2次排序
•原理
•Map之前的数据
         key1  1
         key2  2
         key2  3
         key3  4
         key1  2
•Mapduce只能排序key,所以为了二次排序我们要重新定义自己的key 简单说来就是<key value> value ,组合完后
         <key1  1 >    1
         <key2  2 >    2
         <key2  3 >    3
         <key3  4>     4
         <key1  2 >    2

•原理
•接下来实现自定义的排序类,分组类,数据变成
         <key1  1 >    1
         <key1  2 >    2
         <key2  2 >    2
         <key2  3 >    3
         <key3  4>     4
•最后 reduce处理后输出结果
           key1  1
           key1  2
           key2  2
           key2  3
           key3  4



第四部分:计数器
•什么是计数器
            计数器主要用来收集系统信息和作业运行信息,用于知道作业成功、失败等情况,比日志更便利进行分析。
•内置计数器
•Hadoop内置的计数器,记录作业执行情况和记录情况。包括MapReduce框架、文件系统、作业计数三大类。
•计数器由关联任务维护,定期传递给tasktracker,再由tasktracker传给jobtracker。
•计数器可以被全局聚集。内置的作业计数器实际上由jobtracker维护,不必在整个网络中传递。
•当一个作业执行成功后,计数器的值才是完整可靠的。


用户自定义Java计数器
•MapReduce框架允许用户自定义计数器
•计数器是全局使用的
•计数器有组的概念,可以由一个Java枚举类型来定义
•如何配置
•0.20.2以下的版本使用Reporter,
•0.20.2以上的版本使用context.getCounter(groupName, counterName) 来获取计数器配置并设置。
•动态计数器
•所谓动态计数器即不采用Java枚举的方式来定义

•Reporter中的获取动态计数器的方法
•public void incrCounter(String group,String counter,long amount)
            组名称,计数器名称,计数值

•一些原则
•创建计数器时,尽量让名称易读


•获取计数器
•Web UI
•命令行 hadoop job-counter
•Java API
•Java API
•在作业运行完成后,计数器稳定后获取。 使用job.getCounters()得到Counters



第五部分:合并小文件示例
•产生背景
•Hadoop不适合处理小文件
•会占用大量的内存空间
•解决方案
•文件内容读取到SequenceFile内

已有 0 人发表留言,猛击->> 这里<<-参与讨论


ITeye推荐



相关 [hadoop mapreduce 编程] 推荐:

Hadoop MapReduce高级编程

- - 互联网 - ITeye博客
•combine函数把一个map函数产生的对(多个key, value)合并成一个新的. 将新的作为输入到reduce函数中,其格式与reduce函数相同. •这样可以有效的较少中间结果,减少网络传输负荷. •什么情况下可以使用Combiner.

Hadoop MapReduce编程入门案例

- - CSDN博客云计算推荐文章
Hadoop入门例程简析中. (下面的程序下载地址: http://download.csdn.net/detail/zpcandzhj/7810829). (1)Hadoop新旧API的区别. 新的API倾向于使用虚类(抽象类),而不是接口,因为这更容易扩展. 例如,可以无需修改类的实现而在虚类中添加一个方法(即用默认的实现).

Hadoop MapReduce技巧

- - 简单文本
我在使用Hadoop编写MapReduce程序时,遇到了一些问题,通过在Google上查询资料,并结合自己对Hadoop的理解,逐一解决了这些问题. Hadoop对MapReduce中Key与Value的类型是有要求的,简单说来,这些类型必须支持Hadoop的序列化. 为了提高序列化的性能,Hadoop还为Java中常见的基本类型提供了相应地支持序列化的类型,如IntWritable,LongWritable,并为String类型提供了Text类型.

下一代Hadoop MapReduce

- Jia - NoSQLFan
本文来自Hadoop Summit大会的一个演讲稿,主讲是Hadoop核心开发团队的Arun C Murthy (@acmurthy),同时他也是Yahoo!刚刚剥离的Hadoop独立公司Hortonworks的 Founder和架构师. 演讲中他讲述了现在的Hadoop存在的一些问题和集群上限,并展望了下一代Hadoop和其MapReduce将会得到的巨大提升.

"Hadoop/MapReduce/HBase"分享总结

- - ITeye博客
此分享是关于hadoop生态系统的简单介绍包括起源到相对应用. Hadoop和HBase.pdf (2.1 MB). 已有 0 人发表留言,猛击->> 这里<<-参与讨论. —软件人才免语言低担保 赴美带薪读研.

Hadoop之MapReduce单元测试

- - ITeye博客
通常情况下,我们需要用小数据集来单元测试我们写好的map函数和reduce函数. 而一般我们可以使用Mockito框架来模拟OutputCollector对象(Hadoop版本号小于0.20.0)和Context对象(大于等于0.20.0). 下面是一个简单的WordCount例子:(使用的是新API).

MapReduce编程模型

- - CSDN博客云计算推荐文章
MapReduce是一个Google发明的编程模型,也是一个处理和生成超大规模数据集的算法模型的相关实现. 用户首先创建一个Map函数处理一个基于对的数据集合,输出的中间结果基于对的数据集合,然后再创建一个Reduce函数用来合并所有的具有相同中间Key值的中间Value值.

[转]基于mapreduce的Hadoop join实现

- -
对于一个大数据的分析应用,join是必不可少的一项功能.现在很多构建与hadoop之上的应用,如Hive,PIG等在其内部实现了join程序,可以通过很简单的sql语句或者数据操控脚本完成相应的Join工作.那么join应该如何实现呢?今天我们就对join做一个简单的实现. 我们来看一个例子,现在有两组数据:一组为单位人员信息,如下:.

【Hadoop】MapReduce使用combiner优化性能

- - CSDN博客云计算推荐文章
当MapReduce模型中,reduce执行的任务为统计分类类型的值总量或去重后的数量,或最大值最小值时,可以考虑在Map输出后进行combine操作;这样可以减少网络传输带来的开销,同时减轻了reduce任务的负担. Combine操作是运行在每个节点上的,只会影响本地Map的输出结果;Combine的输入为本地map的输出结果(一般是数据在溢出到磁盘之前,可以减少IO开销),其输出则作为reduce的输入.

hadoop的IO和MapReduce优化参数

- - CSDN博客系统运维推荐文章
           在MapReduce执行过程中,特别是Shuffle阶段,尽量使用内存缓冲区存储数据,减少磁盘溢写次数;同时在作业执行过程中增加并行度,都能够显著提高系统性能,这也是配置优化的一个重要依据.            下面分别介绍I/O属性和MapReduce属性这两个类的部分属性,并指明其优化方向.