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输出的位置,直到它获得所有输出位置.
相关推荐
4 分别在自编 MapReduce 程序 WordCount 运行过程中和运行结束后查看 MapReduce Web 界面。 5. 分别在自编 MapReduce 程序 WordCount 运行过程中和运行结束后练习 MapReduce Shell 常用命令。 。。
(2)打开网站localhost:8088和localhost:50070,查看MapReduce任务启动情况 (3)写wordcount代码并把代码生成jar包 (4)运行命令 (1):把linus下的文件放到hdfs上 (2):运行MapReduce (5):查看运行结果 ...
MapReduce 初级编程实践 姓名: 实验环境: 操作系统:Linux(建议Ubuntu16.04); Hadoop版本:3.2.2; 实验内容与完成情况: (一)编程实现文件合并和去重操作 对于两个输入文件,即文件 A 和文件 B,请...
赠送jar包:hadoop-mapreduce-client-jobclient-2.6.5.jar; 赠送原API文档:hadoop-mapreduce-client-jobclient-2.6.5-javadoc.jar; 赠送源代码:hadoop-mapreduce-client-jobclient-2.6.5-sources.jar; 赠送...
Mapreduce实验报告 前言和简介 MapReduce是Google提出的一种编程模型,在这个模型的支持下可以实现大规模并行化计 算。在Mapreduce框架下一个计算机群通过统一的任务调度将一个巨型任务分成许多部分 ,分别解决然后...
提出一种基于MapReduce技术的贝叶斯垃圾邮件过滤机制,一方面对传统贝叶斯过滤技术进行改进,另一方面利用MapReduce模型的海量数据处理优势优化邮件集训练与学习。实验,较之目前流行的传统贝叶斯算法、K最近邻(NN算法...
【MapReduce篇07】MapReduce之数据清洗ETL1
MapReduce是一种编程模型,用于大规模数据集(大于1TB)的并行运算。概念"Map(映射)"和"Reduce(归约)",是它们的主要思想,都是从函数式编程语言里借来的,还有从矢量编程语言里借来的特性...
项目的具体内容如下:1:用MapReduce算法实现贝叶斯分类器的训练过程,并输出训练模型; 2:用输出的模型对测试集文档进行分类测试。测试过程可基于单机Java程序,也可以是MapReduce程序。输出每个测试文档的分类...
简单的在MapReduce中实现两个表的join连接简单的在MapReduce中实现两个表的join连接简单的在MapReduce中实现两个表的join连接
本节介绍如何编写基本的 MapReduce 程序实现数据分析。本节代码是基于 Hadoop 2.7.3 开发的。 任务准备 单词计数(WordCount)的任务是对一组输入文档中的单词进行分别计数。假设文件的量比较大,每个文档又包含...
Hadoop 用mapreduce实现Wordcount实例,绝对能用
单词计数是最简单也是最能体现 MapReduce 思想的程序之一,可以称为 MapReduce 版“Hello World”。单词计数的主要功能是统计一系列文本文件中每个单词出现的次数。本节通过单词计数实例来阐述采用 MapReduce 解决...
MapReduce求取行平均值 MapReduce小实例 数据有经过处理已经添加行号的 也有未添加的 行平均值的四种求法
MapReduce is the distribution system that the Hadoop MapReduce engine uses to distribute work around a cluster by working parallel on smaller data sets. It is useful in a wide range of applications, ...
《MapReduce2.0源码分析与编程实战》比较系统地介绍了新一代MapReduce2.0的理论体系、架构和程序设计方法。全书分为10章,系统地介绍了HDFS存储系统,Hadoop的文件I/O系统,MapReduce2.0的框架结构和源码分析,...
一个自己写的Hadoop MapReduce实例源码,网上看到不少网友在学习MapReduce编程,但是除了wordcount范例外实例比较少,故上传自己的一个。包含完整实例源码,编译配置文件,测试数据,可执行jar文件,执行脚本及操作...
MapReduce的重要概念 什么是MapReduce 1 MapReduce核心思想 2 MapReduce特点 3 MapReduce适用的开发场景 4 目 录 一、什么是MapReduce MapReduce起源,在介绍大数据编年史时有提到Google最早在04年发表论文MapReduce...
赠送jar包:hadoop-mapreduce-client-common-2.7.3.jar; 赠送原API文档:hadoop-mapreduce-client-common-2.7.3-javadoc.jar; 赠送源代码:hadoop-mapreduce-client-common-2.7.3-sources.jar; 赠送Maven依赖信息...
在hadoop平台上,用mapreduce编程实现大数据的词频统计