大数据开发MapReduce介绍

杰瑞教育    2018-5-13

1.1MapReduce的作用

假设有一个计算文件中单词个数的需求,文件比较多也比较大,在单击运行的时候机器的内存受限,磁盘受限,运算能力受限,而一旦将单机版程序扩展到集群来分布式运行,将极大增加程序的复杂度和开发难度,因此这个工作可能完成不了。针对以上这个案例,MapReduce在这里能起到什么作用呢,引入MapReduce框架后,开发人员可以将绝大部分工作集中在业务逻辑的开发上,而将分布式计算中的复杂性交由框架来处理。

可见在程序由单机版扩成分布式时,会引入大量的复杂工作。为了提高开发效率,可以将分布式程序中的公共功能封装成框架,让开发人员可以将精力集中于业务逻辑。而MapReduce就是这样一个分布式程序的通用框架。

 

1.2MapReduce架构图

 

MapReduce也采用了Master/Slave(M/S)架构。它主要由以下几个组件组成 :Client、JobTracker、TaskTracker和Task。下面分别对这几个组件进行介绍。

(1)Client

 用户编写的MapReduce程序通过Client提交到JobTracker端;同时用户可通过Client提供的一些接口查看作业运行状态。在Hadoop内部用“作业”(Job)表示MapReduce程序。一个MapReduce程序可对应若干个作业,而每个作业会被分解成若干个Map/Reduce任务(Task)。

(2)JobTracker

JobTracker主要负责资源监控和作业调度。JobTracker监控所有TaskTracker与作业Job的健康状况,一旦发现失败情况后,其会将相应的任务转移到其他节点;同时,JobTracker会跟踪任务的执行进度、资源使用量等信息,并将这些信息告诉任务调度器,而调度器会在资源出现空闲时,选择合适的任务使用这些资源。在Hadoop中,任务调度器是一个可插拔的模块,用户可以根据自己的需要设计相应的调度器。

(3)TaskTracker

TaskTracker会周期性地通过Heartbeat将本节点上资源的使用情况和任务的运行进度汇报给JobTracker,同时接收JobTracker发送过来的命令并执行相应的操作(如启动新任务、杀死 任务等)。TaskTracker使用“slot”等量划分本节点上的资源量。“slot”代表计算资源(CPU、 内存等)。一个Task获取到一个slot后才有机会运行,而Hadoop调度器的作用就是将各个TaskTracker上的空闲slot分配给Task使用。slot分为Map slot和Reduce slot两种,分别供Map Task和Reduce Task使用。TaskTracker通过slot数目(可配置参数)限定Task的并发度。

(4)Task

Task分为Map Task和Reduce Task两种,均由TaskTracker启动。从上一小节中我们知道,HDFS以固定大小的block为基本单位存储数据,而对于MapReduce而言,其处理单位是split。split与block的对应关系如下图所示。split是一个逻辑概念,它只包含一些元数据信息,比如 数据起始位置、数据长度、数据所在节点等。它的划分方法完全由用户自己决定。但需要注意的是,split的多少决定了Map Task的数目,因为每个split会交由一个Map Task处理。

 

Map Task执行过程如下图所示。由该图可知,Map Task先将对应的split迭代解析成一 个个key/value对,依次调用用户自定义的map()函数进行处理,最终将临时结果存放到本地磁盘上,其中临时数据被分成若干个partition(分片),每个partition将被一个Reduce Task处理。

Reduce Task执行过程如下图所示。该过程分为三个阶段:

①从远程节点上读取Map Task中间结果(称为“Shuffle阶段”);

②按照key对key/value对进行排序(称为“Sort阶段”);

③依次读取<key, value list>,调用用户自定义的reduce()函数处理,并将最终结果存到HDFS上(称为“Reduce阶段”)。

MapReduce是一种并行编程模式,利用这种模式软件开发者可以轻松地编写出分布式并行程序。在Hadoop的体系结构中,MapReduce是一个简单易用的软件框架,基于它可以将任务分发到由上千台商用机器组成的集群上,并以一种可靠容错的方式并行处理大量的数据集,实现Hadoop的并行任务处理功能。MapReduce框架是由一个单独运行在主节点的JobTrack和运行在每个集群从节点的TaskTrack共同组成的。

主节点负责调度构成一个作业的所有任务,这些任务分布在不同的节点上。主节点监控它们的执行情况,并且重新执行之前失败的任务;

从节点仅负责由主节点指派的任务。

当一个Job任务被提交时,JobTrack接收到提交作业和其配置信息之后,就会配置信息等发给从节点,同时调度任务并监控TaskTrack的执行。

 

1.3MapReduce程序运行演示

Hadoop的发布包中内置了一个hadoop-mapreduce-example-2.6.5.jar,这个jar包中有各种MR示例程序,可以通过以下步骤运行:

启动hdfs,yarn,然后在集群中的任意一台服务器上启动执行程序(比如运行wordcount):

hadoop jar hadoop-mapreduce-example-2.6.5.jar wordcount  /wordcount/data /wordcount/out

 

2.1编程规范

 

1) 用户编写的程序分成三个部分:Mapper,Reducer,Driver(提交运行mr程序的客户端)

2) Mapper的输入数据是KV对的形式(KV的类型可自定义)

3) Mapper的输出数据是KV对的形式(KV的类型可自定义)

4) Mapper中的业务逻辑写在map()方法中

5) map()方法(maptask进程)对每一个<K,V>调用一次

6) Reducer的输入数据类型对应Mapper的输出数据类型,也是KV

7) Reducer的业务逻辑写在reduce()方法中

8) Reducetask进程对每一组相同k的<k,v>组调用一次reduce()方法

9) 用户自定义的Mapper和Reducer都要继承各自的父类

10) 整个程序需要一个Drvier来进行提交,提交的是一个描述了各种必要信息的job对象

 

2.2wordcount 示例编写

 

需求:在一堆给定的文本文件中统计输出每一个单词出现的总次数

(1)定义一个mapper类


//首先要定义四个泛型的类型//keyin:  LongWritable    valuein: Text//keyout: Text            valueout:IntWritable

publicclassWordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{//map方法的生命周期:  框架每传一行数据就被调用一次//key :  这一行的起始点在文件中的偏移量//value: 这一行的内容
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//拿到一行数据转换为string
        String line=value.toString();//将这一行切分出各个单词
        String[] words= line.split("");//遍历数组,输出<单词,1>for(String word:words){
            context.write(new Text(word), new IntWritable(1));
        }
    }
}

(2)定义一个reducer类


//生命周期:框架每传递进来一个kv 组,reduce方法被调用一次
    @Override
    protected void reduce(Text key, Iterable<IntWritable>values, Context context) throws IOException, InterruptedException {//定义一个计数器
        int count=0;//遍历这一组kv的所有v,累加到count中for(IntWritable value:values){
            count+=value.get();
        }
        context.write(key, new IntWritable(count));
    }
}

(3)定义一个主类,用来描述job并提交job


publicclassWordCountRunner {//把业务逻辑相关的信息(哪个是mapper,哪个是reducer,要处理的数据在哪里,输出的结果放哪里……)描述成一个job对象//把这个描述好的job提交给集群去运行
    public static void main(String[] args) throws Exception {
        Configuration conf=new Configuration();
        Job wcjob=Job.getInstance(conf);//指定我这个job所在的jar包//        wcjob.setJar("/home/hadoop/wordcount.jar");
        wcjob.setJarByClass(WordCountRunner.class);
        
        wcjob.setMapperClass(WordCountMapper.class);
        wcjob.setReducerClass(WordCountReducer.class);//设置我们的业务逻辑Mapper类的输出key和value的数据类型
        wcjob.setMapOutputKeyClass(Text.class);
        wcjob.setMapOutputValueClass(IntWritable.class);//设置我们的业务逻辑Reducer类的输出key和value的数据类型
        wcjob.setOutputKeyClass(Text.class);
        wcjob.setOutputValueClass(IntWritable.class);//指定要处理的数据所在的位置
        FileInputFormat.setInputPaths(wcjob,"hdfs://hdp-server01:9000/wordcount/data/big.txt");//指定处理完成之后的结果所保存的位置
        FileOutputFormat.setOutputPath(wcjob, new Path("hdfs://hdp-server01:9000/wordcount/output/"));//向yarn集群提交这个job
        boolean res=wcjob.waitForCompletion(true);
        System.exit(res?0:1);
    }

2.3集群运行模式

 

1) 将mapreduce程序提交给yarn集群resourcemanager,分发到很多的节点上并发执行

2) 处理的数据和输出结果应该位于hdfs文件系统

3) 提交集群的实现步骤:

将程序打成JAR包,然后在集群的任意一个节点上用hadoop命令启动hadoop jar wordcount.jar cn.bigdata.mrsimple.WordCountDriver inputpath outputpath

 

作者:杰瑞教育
出处:http://www.cnblogs.com/jerehedu/ 
版权声明:本文版权归烟台杰瑞教育科技有限公司和博客园共有,欢迎转载,但未经作者同意必须保留此段声明,且在文章页面明显位置给出原文连接,否则保留追究法律责任的权利。 





暗色背景
手机扫码阅读本页