文档章节

hadoop两大核心之一:MapReduce总结

迪allen
 迪allen
发布于 2014/03/27 00:06
字数 2329
阅读 2579
收藏 4

MapReduce

         MapReduce是一种分布式计算模型,由Google提出,主要用于搜索领域,MapReduce程序

        本质上是并行运行的,因此可以解决海量数据的计算问题.

         MapReduce任务过程被分为两个处理阶段:map阶段和reduce阶段.每个阶段都以键

         值对作为输入和输出.用户只需要实现map()和reduce()两个函数即可实现分布式计算.

     执行步骤:

         map任务处理:

             1.读取输入文件内容,解析成键值对(key/value).对输入文件的每一行,解析成

                键值对(key/value).每一个键值对调用一次map函数

             2.写自己的逻辑,对输入的键值对(key/value)处理,转换成新的键值对

                (key/value)输出.

             3.对输出的键值对(key/value)进行分区.(partition)

             4.对不同分区的数据,按照key进行排序,分组.相同的key/value放到

                一个集合中.(shuffle)

             5.分组后的数据进行规约.(combiner,可选择的)

     reduce任务处理:

         1.对多个map任务的输出,按照不同的分区,通过网络copy到不同的reduce节点.

         2.对多个map任务的输出进行合并,排序.写reduce函数自己的逻辑,对输入的

            key/value处理,转换成新的key/value输出.

         3.把reduce的输出保存到文件中(写入到hdfs中).

     MapReduce作业流程:

         1.代码编写

         2.作业配置(输入输出路径,reduce数量等等)

         3.作业提交

             3.1通过JobClient提交,与JobTracker通信得到一个jar的存储路径和JobId.

             3.2检查输入输出的路径

             3.3计算分片信息.

             3.4将作于所需要的资源(jar,配置文件,计算所得的输入分片)赋值到以作业ID

                  命名的HDFS上

             3.5告知JobTracker作业准备执行.

         4.作业初始化

             当JobTracker接收到提交过来的作业后,会把次调用放入一个内部队列中,

             交由作业调度器进行调度,默认是(FIFO),并对其初始化.

             初始化:创建一个表示正在运行作业的对象--分装任务和记录信息,以便

             跟踪任务的状态和进程.

             为了创建任务列表,作业调度器首先从共享文件系统中获取已经计算好的

             输入分片信息.然后为每一个分片创建一个map任务,调度器创建相应数量

             的要运行的reduce任务.此时,任务被指定ID.

         5.任务分配

             tasktracker运行一个简单的循环来定期发送"心跳"给JobTracker,

             心跳告知JobTracker,tasktracker是否还存活,同时指明tasktracker是否

             已经准备好运行新的任务,如果是,JobTracker会分配给它一个任务.

         6.任务执行

             tasktracker拿到任务后

             1.会将所有的信息拷贝到本地(包括jar,代码,配置信息,分片信息等)

             2.tasktracker为任务新建一个本地工作目录,并把jar文件中的内容解压到

                这个目录下.

             3.tasktracker新建一个TaskRunner实例来运行该任务.

              TaskRunner会启动一个新的JVM来运行每个步骤.(防止其他软件影响

                到tasktracker,但是在不同的任务之间重用JVM是有可能的.

         7.进度和状态的更新

             task会定期向tasktracker汇报执行情况,tasktracker会定期收集所在

             集群上的所有task信息,并想JobTracker汇报.JobTracker会根据所有

             tasktracker汇报上来的信息进行汇总

         8.作业完成

             JobTracker是在接收到最后一个任务完成后,才将任务标记为"成功".

            并将数据结果写入到HDFS上.

         PS:

             JobTracker职能:负责接收用户提交的作业,负责启动,跟踪任务执行

             tasktracker职能:负责执行任务

     作业失败:

         1.JobTracker失败

             这是最为严重的一种任务失败,失败机制--它是一个单节点故障,因此,

             作业注定失败.(hadoop2.0解决了)

         2.tasktracker失败

             tasktracker崩溃了会停止向jobt发送心跳信息,并且JobTracker会将

             tasktracker从等待的任务池中移除,将该任务转移到其他的地方执行.

             JobTracker会将tasktracker加入到黑名单.

         3.task失败

             map或reduce运行失败,会向tasktracker抛出异常,任务挂起.

MapReduce启动流程:

             start-mapred.sh  --> hadoop-daemon.sh --> hadoop 

            -->org.apache.hadoop.mapred.JobTracker

    Jobtracker调用顺序:

             main --> startTracker  --> new JobTracker 在其构造方法中首先创建

             一个调度器,接着创建一个RPC的server(interTrackerServer)

             tasktracker会通过PRC接触与其通信, 然后调用offerService方法对外

             提供服务,在offerService方法中启动RPC server,初始化jobtracker,

             调用taskScheduler的start方法 --> eagerTaskInitializationListener

             调用start方法,接着调用jobInitManagerThread的start方法,

             因为其是一个线程,会调用JobInitManager的run方法,随后

             jobInitQueue任务队列去取第一个任务,然后把它丢入线程池中,

            再调用-->InitJob的run方法,再然后调用jobTracker的initJob方法

             --> JobInProgress的initTasks 

             --> maps = new TaskInProgress[numMapTasks]和

             reduces = new TaskInProgress[numReduceTasks];

     TaskTracker调用顺序:

             main --> new TaskTracker在其构造方法中调用了initialize方法,

             在initialize方法中调用RPC.waitForProxy,得到一个jobtracker的

             代理对象,接着TaskTracker调用了本身的run方法,

             --> offerService方法  --> transmitHeartBeat返回值是

              (HeartbeatResponse)是jobTracker的指令,在transmitHeartBeat

             方法中InterTrackerProtocol调用了heartbeat将tasktracker的状态

             通过RPC机制发送给jobTracker,返回值就是JobTracker的指

             令,heartbeatResponse.getActions()得到具体的指令,然后判断指令

             的具体类型,开始执行任务,addToTaskQueue动类型的指令

             加入到队列当中,TaskLauncher又把任务加入到任务队列当中,

               -->  TaskLauncher的run方法 --> startNewTask方法 

             --> localizeJob下载资源 --> launchTaskForJob开始加载任务

             --> launchTask  --> runner.start()启动线程; --> 

            TaskRunner调用run方法 --> launchJvmAndWait启动java child进程

 

MapReduce的细节

     序列化概念

         序列化:是指把结构化对象转化为字节流.

         反序列化:是序列化的逆过程.即:把字节流转回结构化对象.

     hadoop序列化格式特点:

         1.紧凑:高效使用存储空间

         2.快速:读写数据的额外开销小

         3.可扩展:可透明的读取老格式的数据

         4.互操作:支持多语言交互.

     hadoop序列化的作用:

         序列化在分布式环境的两大作用:进程间的通信,永久存储.

         hadoop节点间通信.

Partitioner编程

    将有一些共同特性的数据,写入到同一个文件里.

排序和分组

     在map和reduce阶段进行排序时,比较的是k2。v2是不参与排序比较的。

     如果要想让v2也进行排序,需要把k2和v2组装成新的类,作为k2,

     才能参与比较。如果想自定义排序规则,被排序的对象要实现

     WritableComparable接口,在compareTo方法中实现排序规则,

     然后将这个对象当做k2,即可完成排序分组时也是按照k2进行比较的。

Combiners编程

     1.每一个map会产生大量的输出,combiner的作用就是在map端对输出

       先做一次合并,以减少传输到reducer的数据量.

     2.combiner最基本是实现本地key的归并,具有类似本地reduce功能

      如果不用combiner,那么所有的结果都是reduce完成,效率会相对降低,

     3.使用combiner,先完成的map会在本地聚合,提升速度.

     PS:Combiner的输出是Reducer的输入,Combiner绝对不能改变最终的计算结果.

       所以从个人角度来说,Combiner只应用于那种Reducer的输入key/value与

        输出key/value类型完全一致,且不影响最终结果的场景.比如:累加,最值等.

  

shuffle

         MapReduce确保每个Reducer的输入都按键排序,系统执行排序的过程

         ---将map输出作为输入传给Reducer---成为shuffle(洗牌)

         1.map函数开始产生输出时,并不是简单的将它写到磁盘.它利用缓冲的

            方式写到内存,并出于效率的考虑进行了预排序.

          每个map任务都有一个环形内存缓冲区,用于存储任务的输出.默认情况下,

            缓冲区大小为100MB,一旦缓冲内容达到阀值(默认为80%),一个后台线程

            便开始把内容写到磁盘指定目录下新建一个溢写文件中.在写到磁盘过程中,

            map输出继续被写到缓冲区,但如果在此期间缓冲区被填满, map会阻塞,

            直到写磁盘过程完成.

         2.写磁盘前,要partition,sort。如果有combiner,combine排序后数据。

         3.等最后记录写完,合并全部溢写文件为一个分区且排序的文件。

        

    Reducer如何知道要从哪个tasktracker取得map输出呢?

         map任务成功完成后,它们会通知其父tasktracker状态已更新,然后tasktracker

        进而通知JobTracker.这些通知在心跳机制中传输.因此,对于指定作业,

        JobTracker知道map输出和tasktracker之间的映射关系. Reducer中的一个

        线程定期询问JobTracker以便获取map输出的位置,直到它获得所有输出位置.


© 著作权归作者所有

迪allen
粉丝 28
博文 14
码字总数 11554
作品 0
程序员
私信 提问
大数据hadoop是什么?初学者如何进行系统学习?

大数据的火爆程度在不断的增加,似乎一个行业不和大数据相对接就会显得很“LOW”。大数据行业的薪资水平越来越高,决定了更多的人纷纷学习大数据,hadoop作为大数据的一个核心自然就是学习的...

hainiubuluo
2018/05/25
0
0
Spark VS Hadoop 两大大数据分析系统深度解读

大数据,无论是从产业上,还是从技术上来看,都是目前的发展热点。在中国,政府控制着80%的数据,剩下的多由“BAT”这样的大公司拥有,中小企业如何构建自己的大数据系统?其他企业如何建设自...

景龙Edward
2016/06/16
11.4K
16
Hadoop的辉煌还能延续多久?

Hadoop技术已经无处不在。不管是好是坏,Hadoop已经成为大数据的代名词。短短几年间,Hadoop从一种边缘技术成为事实上的标准。看来,不仅现在Hadoop是企业大数据的标准,而且在未来,它的地位...

一枚Sir
2014/08/05
246
0
大数据教程(8.1)mapreduce核心思想

上一章介绍了hadoop的HDFS文件系统的原理及API使用。本章博主将继续对hadoop的mapreduce编程框架进行分享。 mapreduce原理篇 mapreduce是一个分布式运算程序的编程框架,是用户开发“基于had...

em_aaron
2018/11/19
124
0
Hadoop学习笔记(1):概念和整体架构

hadoop简介和历史   Hadoop架构体系   Master和Slave节点   数据分析面临的问题和Hadoop思想   由于工作原因,必须学习和深入一下Hadoop,特此记录笔记。   什么是hadoop?   Apa...

ugali
2018/06/26
0
0

没有更多内容

加载失败,请刷新页面

加载更多

Raspberry Pi 树莓派 搭建java8环境

更新软件源 apt-get upgradeapt-get update 移除自带的openjdk(如果有的话) apt-get remove openjdk* 安装oracle jdk,地址: https://www.oracle.com/technetwork/java/javase/downl......

ChangeZ
今天
79
0
树莓派3b 安装openwrt

https://downloads.openwrt.org/releases/19.07.0/targets/brcm2708/bcm2710/openwrt-19.07.0-brcm2708-bcm2710-rpi-3-squashfs-factory.img.gz 插入并启动树莓派3B+,顺便接上键盘和显示器,......

FalconChen
今天
71
0
OSChina 周一乱弹 —— 有些运动也常用到膝盖

Osc乱弹歌单(2020)请戳(这里) 【今日歌曲】 @薛定谔的兄弟 :分享洛神有语创建的歌单「我喜欢的音乐」: 《火宵の月・テーマ~ピアノソロ》- 中村由利子 手机党少年们想听歌,请使劲儿戳(...

小小编辑
今天
109
0
Redis持久化机制

RDB存储 RDB方式的持久化是通过快照完成的,当符合一定条件时Redis会自动将内存中的所有数据生成一份副本并存储在硬盘上。 Redis会在一下集中情况下对数据进行快照: 根据配置规则进行快照;...

XuePeng77
今天
81
0
float精度计算测试

package mainimport ("fmt""github.com/go-ozzo/ozzo-dbx"_ "github.com/go-sql-driver/mysql")type DecimalDemo struct {Id intAmount float64}func main() {db, _......

醉人的笑容你有没有
今天
127
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部