文档章节

Hadoop学习笔记——高级编程

超人学院
 超人学院
发布于 2014/11/25 11:26
字数 2241
阅读 191
收藏 2

从前面的学习中,我们了解到了MapReduce整个过程需要经过以下几个步骤:

1.输入(input):将输入数据分成一个个split,并将split进一步拆成<key, value>。

2.映射(map):根据输入的<key, value>进生处理,

3.合并(combiner):合并中间相两同的key值。

4.分区(Partition):将<key, value>分成N分,分别送到下一环节。

5.化简(Reduce):将中间结果合并,得到最终结果

6.输出(output):负责输入最终结果。

其中第3、4步又成洗牌(shuffle)过程。

 

从前面HelloWorld示例中,我们看到,我们只去个性化了Map和Reduce函数,那其他函数呢,是否可以个性化?答案当然是肯定的。下面我们就对每个环节的个性化进行介绍。

自定义输入格式

输入格式(InputFormat)用于描述整个MapReduce作业的数据输入规范。先对输入的文件进行格式规范检查,如输入路径,后缀等检查;然后对数据文件进行输入分块(split);再对数据块逐一读出;最后转换成Map所需要的<key, value>健值对。

系统中提供丰富的预置输入格式。最常用的以下两种:

TextInputFormat:系统默认的数据输入格式。将文件分块,并逐行读入,每一行记录行成一对<key, value>。其中,key值为当前行在整个文件中的偏移量,value值为这一行的文本内容。

KeyValueTextInputFormat:这是另一个常用的数据输入格式,读入的文本文件内容要求是以<key, value>形式。读出的结果也就直接形成<key, value>送入map函数中。

 

如果选择输入格式呢?那就只要在job函数中调用

  1. job.setInputFormatClass(TextInputFormat.class);

在Hello中我们没有设定,系统默认选择了TextInputFormat。

一般情况够用了,但某些情况下,还是无法满足用户的需求,所以还是需要个性化。个性化则按下面的方式进行:

如果数据我们是来源于文件,则可以继承FileInputFormat:

  1. public class MyInputFormat extends FileInputFormat<Text,Text> {

  2.    @Override

  3.    public RecordReader<Text, Text> createRecordReader(InputSplit split,

  4.          TaskAttemptContext context) throws IOException, InterruptedException {

  5.       // TODO Auto-generated method stub

  6.       return null;

  7.    }

  8. }

如果数据我们是来源于非文件,如关系数据,则继承

  1. public class MyInputFormat extends InputFormat<Text,Text> {

  2.  

  3.    @Override

  4.    public RecordReader<Text, Text> createRecordReader(InputSplit arg0,

  5.          TaskAttemptContext arg1) throws IOException, InterruptedException {

  6.       // TODO Auto-generated method stub

  7.       return null;

  8.    }

  9.  

  10.    @Override

  11.    public List<InputSplit> getSplits(JobContext arg0) throws IOException,

  12.          InterruptedException {

  13.       // TODO Auto-generated method stub

  14.       return null;

  15.    }

  16.  

  17. }

这里比较清晰了,下面个函数为拆分成split,上面个函数跟据split输出成Key,value。

 

自定义map处理

这个好理解,我们的HelloWorld程序中就自定义了map处理函数。然后在job中指定了我们的处理类:

  1. job.setMapperClass(TokenizerMapper.class);

能不能没有map呢? 可以的,如果没有map,也就是这与上面的这个setMapperClass,则系统自动指定一个null,这时处理是将输入的<key,value>值,不作任何修改,直接送到下一环节中。

个性化代码如下:

  1. public static class TokenizerMapper

  2.        extends Mapper<Object, Text, Text, IntWritable>{

  3.  

  4.     public void map(Object key, Text value, Context context

  5.                     ) throws IOException, InterruptedException {

  6.  

  7.         context.write(key, value);

  8.     }

  9.   }

 

自定义合并Combiner

自定义合并Combiner类,主要目的是减少Map阶段输出中间结果的数据量,降低数据的网络传输开销。

Combine过程,实际跟Reduce过程相似,只是执行不同,Reduce是在Reducer环节运行,而Combine是紧跟着Map之后,在同一台机器上预先将结时进行一轮合并,以减少送到Reducer的数据量。所以在HelloWorld时,可以看到,Combiner和Reducer用的是同一个类:

  1. job.setCombinerClass(IntSumReducer.class);

  2. job.setReducerClass(IntSumReducer.class);

如何个性化呢,这个跟Reducer差不多了:

  1. public static class MyCombiner

  2.       extends Reducer<Text,IntWritable,Text,IntWritable> {

  3.  

  4.    public void reduce(Text key, Iterable<IntWritable> values,

  5.                       Context context

  6.                       ) throws IOException, InterruptedException {

  7.  

  8.      context.write(key, new IntWritable(1));

  9.    }

  10.  }

 

自定义分区Partitioner

在MapReduce程序中,Partitioner决定着Map节点的输出将被分区到哪个Reduce节点。而默认的Partitioner是HashPartitioner,它根据每条数据记录的主健值进行Hash操作,获得一个非负整数的Hash码,然后用当前作业的Reduce节点数取模运算,有N个结点的话,就会平均分配置到N个节点上,一个隔一个依次。大多情况下这个平均分配是够用了,但也会有一些特殊情况,比如某个文件的,不能被拆开到两个结点中,这样就需要个性化了。

个性化方式如下:

  1. public static class MyPartitioner

  2.       extends HashPartitioner<K,V> {

  3.  

  4.    public void getPartition(K key, V value,int numReduceTasks) {

  5.  

  6.      super.getPartition(key,value,numReduceTasks);

  7.    }

  8.  }

方式其实就是在执行之前可以改变一下key,来欺骗这个hash表。

 

自定义化简(Reducer)

这一块是将Map送来的结果进行化简处理,并形成最终的输出值。与前面map一样,在HelloWorld中我们就见到过了。通过下面代码可以设置其值:

  1. job.setReducerClass(IntSumReducer.class);

同样,也可以这样类可以不设置,如果不设置的话,就是把前面送来的值,直接送向输出格式器中。

如果要个性化,则如下:

  1.   public static class IntSumReducer

  2.      extends Reducer<Text,IntWritable,Text,IntWritable> {

  3.  

  4.   public void reduce(Text key, Iterable<IntWritable> values,

  5.                      Context context

  6.                      ) throws IOException, InterruptedException {

  7.     context.write(key, result);

  8.   }

  9. }

 

自定义输出格式

数据输出格式(OutPutFormat)用于描述MapReduce作业的数据输出规范。Hadoop提供了丰富的内置数据输出格式。最常的数据输出格式是TextOutputFormat,也是系统默认的数据输出格式,将结果以"key+\t+value"的形式逐行输出到文本文件中。还有其它的,如:DBOutputFormat,FileOutputFormat,FilterOutputFormat,IndexUpdataOutputFormat,LazyOutputFormat,MapFileOutputFormat,等等。

如果要个性化,则按下面方式进行:

  1. public class MyOutputFormat extends OutputFormat<Text,Text> {

  2.  

  3.    @Override

  4.    public void checkOutputSpecs(JobContext arg0) throws IOException,

  5.          InterruptedException {

  6.       // TODO Auto-generated method stub

  7.  

  8.    }

  9.  

  10.    @Override

  11.    public OutputCommitter getOutputCommitter(TaskAttemptContext arg0)

  12.          throws IOException, InterruptedException {

  13.       // TODO Auto-generated method stub

  14.       return null;

  15.    }

  16.  

  17.    @Override

  18.    public RecordWriter<Text, Text> getRecordWriter(TaskAttemptContext arg0)

  19.          throws IOException, InterruptedException {

  20.       // TODO Auto-generated method stub

  21.       return null;

  22.    }

  23.  

  24. }

 

复合健——用户自定义类型。

从前面的整个过程中可以看到,都是采用key-value的方式进行传入传出,而这些类型大多是单一的字符串,和整型。如果我的key中需要包含多个信息怎么办?用字符串直接拼接么? 太不方便了,最好能够自己定义一个类,作为这个key,这样就方便了。

如果定义一个类作为key 或value的类型? 有什么要求?就是这个类型必须要继承WritableComparable<T>这个类,所以如果要自定义一个类型则可以这么实现:

  1. public class MyType implements WritableComparable<MyType> {

  2.  

  3.    private float x,y;

  4.    public float GetX(){return x;}

  5.    public float GetY(){return y;}

  6.  

  7.       @Override

  8.       public void readFields(DataInput in) throws IOException {

  9.          x = in.readFloat();

  10.          y = in.readFloat();

  11.       }

  12.  

  13.       @Override

  14.       public void write(DataOutput out) throws IOException {

  15.          out.writeFloat(x);

  16.          out.writeFloat(y);

  17.       }

  18.  

  19.       @Override

  20.       public int compareTo(MyType arg0) {

  21.          //输入:-1(小于) 0(等于) 1(大于)

  22.          return 0;

  23.       }

  24.    }

这个示例中,我们添加了两个float变量:x,y 。 这个信息能过int 和out按次序进行输入输出。最后,再实现一个比较函数即可。

 

Job任务的创建

  1. Job job = new Job(conf, "word count");

  2.    job.setJarByClass(WordCount.class);

  3.    job.setInputFormatClass(MyInputFormat.class);

  4.    job.setMapperClass(TokenizerMapper.class);

  5.    job.setCombinerClass(IntSumReducer.class);

  6.    job.setPartitionerClass(MyPartitioner.class);

  7.    job.setReducerClass(IntSumReducer.class);

  8.    job.setOutputFormatClass(TextOutputFormat.class);

  9.    job.setOutputKeyClass(Text.class);

  10.    job.setOutputValueClass(IntWritable.class);

  11.    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

  12.    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

任务创建比较容易,其实就是new一个实例,然后把上面描述的过程类设置好,然后加上第2行中,jar包的主类,第10、11行的输入输出路径。这样就完事了。

 

Job任务的执行

单个任务的执行,没有什么问题,可以用这个:

  1. job.waitForCompletion(true);

但多个任务呢? 多个任务的话,就会形成其组织方式,有串行,有并行,有无关,有组合的,如下图:

图中,Job2和Job3将会等Job1执行完了再执行,且可以同时开始,而Job4必须等Job2和Job3同时结束后才结束。

这个组合,就可以采用这样的代码来实现:

  1. Configuration conf = new Configuration();

  2.       Job job1 = new Job(conf, "job1");

  3.       //.. config Job1

  4.       Job job2 = new Job(conf, "job2");

  5.       //.. config Job2

  6.       Job job3 = new Job(conf, "job3");

  7.       //.. config Job3

  8.       Job job4 = new Job(conf, "job4");

  9.       //.. config Job4

  10.  

  11.       //添加依赖关系

  12.       job2.addDependingJob(job1);

  13.       job3.addDependingJob(job1);

  14.       job4.addDependingJob(job2);

  15.       job4.addDependingJob(job3);

  16.  

  17.       JobControl jc = new JobControl("jbo name");

  18.       jc.addJob(job1);

  19.       jc.addJob(job2);

  20.       jc.addJob(job3);

  21.       jc.addJob(job4);

  22.       jc.run();

 

总述

现在回头看看,其实整个hadoop编程,也就是这几块内容了,要实现某个功能,我们就往上面这些步骤上套,然后联起来执行,达到我们的目的。

更多详情请关注:http://www.superwu.cn Q群:340822685

© 著作权归作者所有

共有 人打赏支持
超人学院
粉丝 110
博文 335
码字总数 388917
作品 0
昌平
CTO(技术副总裁)
私信 提问
Cloudera大数据培训

作为大数据核心技术,Hadoop为企业提供了高扩展、高冗余、高容错、和经济有效的“数据驱动”解决方案,cloudera作为Hadoop的领航者,针对企业目前普遍缺乏海量数据技术人员的现状,开展了Clo...

hadoopstar
2015/03/17
72
0
Golang学习笔记目录

Golang 介绍 Go语言是谷歌2009发布的第二款开源编程语言。 Go语言专门针对多处理器系统应用程序的编程进行了优化,使用Go编译的程序可以媲美C或C++代码的速度,而且更加安全、支持并行进程。...

ChainZhang
2017/12/26
0
0
上海hadoop培训

上海市信息技术培训中心cloudera hadoop管理认证培训 【课程内容】 1、Hadoop分布式文件系统(HDFS) 2、YARN/MapReduce的工作原理 3、如何优化Hadoop机群所需要的硬件配置 4、搭建Hadoop机群...

sittc7
2014/08/14
1K
2
Posix多线程编程学习笔记

Posix多线程编程学习笔记(一)—线程基础(1) http://blog.chinaunix.net/u/22935/showart_310711.html Posix多线程编程学习笔记(一)—线程基础(2) http://blog.chinaunix.net/u/2293...

晨曦之光
2012/04/13
225
0
Spring.NET学习笔记——目录(原)

目录 前言 Spring.NET学习笔记——前言 第一阶段:控制反转与依赖注入IoC&DI Spring.NET学习笔记1——控制反转(基础篇) Level 200 Spring.NET学习笔记2——环境搭建(基础篇) Level 200 Sprin...

长平狐
2012/06/11
913
1

没有更多内容

加载失败,请刷新页面

加载更多

EOS官方钱包keosd

EOS官方钱包的名称是keosd,它负责管理你的私钥,并且帮你进行交易的签名。 不过不幸的是,keosd钱包对普通用户并不友好,它是一个命令行程序,目前还没有像以太坊的mist那样的图形化界面,而...

汇智网教程
今天
28
0
ArrayList的实现原理以及实现线程安全

一、ArrayList概述 ArrayList是基于数组实现的,是一个动态的数字,可以自动扩容。 ArrayList不是线程安全的,效率比较高,只能用于单线程的环境中,在多线程环境中可以使用Collections.syn...

一看就喷亏的小猿
今天
35
0
Netty 备录 (一)

入职新公司不久,修修补补1个月的bug,来了点实战性的技术---基于netty即时通信 还好之前对socket有所使用及了解,入手netty应该不是很难吧,好吧,的确有点难,刚看这玩意的时候,可能都不知道哪里...

_大侠__
昨天
40
0
Django简单介绍和用户访问流程

Python下有许多款不同的 Web 框架。Django是重量级选手中最有代表性的一位。许多成功的网站和APP都基于Django。 Django是一个开放源代码的Web应用框架,由Python写成。 Django遵守BSD版权,初...

枫叶云
昨天
53
0
Spring Cloud Stream消费失败后的处理策略(四):重新入队(RabbitMQ)

应用场景 之前我们已经通过《Spring Cloud Stream消费失败后的处理策略(一):自动重试》一文介绍了Spring Cloud Stream默认的消息重试功能。本文将介绍RabbitMQ的binder提供的另外一种重试...

程序猿DD
昨天
25
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部