Hadoop基础学习---6、MapReduce框架原理(一)

简介: Hadoop基础学习---6、MapReduce框架原理(一)

1、MapReduce框架原理

a40c31fa66bc409a89242742688ddfe8.png

1.1 InputFormat数据输入

1.1.1 切片与MapTask并行度决定机制

1、问题引出

MapTask的并行度决定Map阶段的任务处理并发度,进而影响到整个job的处理速度。

2、MapTask并行度决定机制

数据块:Block是HDFS物理上吧数据分成一块一块。数据块是HDFS储存数据单位。

数据切片:数据切片只是在逻辑上对输出进行分片,并不会在磁盘上将其切分成片进行储存。数据切片是MapReduce程序计算输入数据的单位,一个切片会对应启动一个MapTask。

377520340d814517847a0100f0ff0eff.png

1.1.2 Job提交流程源码与切片源码详解

1、Job提交流程源码解析

waitForCompletion()
submit();
// 1 建立连接
connect();
// 1)创建提交 Job 的代理
new Cluster(getConfiguration());
// (1)判断是本地运行环境还是 yarn 集群运行环境
initialize(jobTrackAddr, conf); 
// 2 提交 job
submitter.submitJobInternal(Job.this, cluster)
// 1)创建给集群提交数据的 Stag 路径
Path jobStagingArea = JobSubmissionFiles.getStagingDir(cluster, conf);
// 2)获取 jobid ,并创建 Job 路径
JobID jobId = submitClient.getNewJobID();
// 3)拷贝 jar 包到集群
copyAndConfigureFiles(job, submitJobDir);
rUploader.uploadFiles(job, jobSubmitDir);
// 4)计算切片,生成切片规划文件
writeSplits(job, submitJobDir);
maps = writeNewSplits(job, jobSubmitDir);
input.getSplits(job);
// 5)向 Stag 路径写 XML 配置文件
writeConf(conf, submitJobFile);
conf.writeXml(out);
// 6)提交 Job,返回提交状态
status = submitClient.submitJob(jobId, submitJobDir.toString(),
job.getCredentials());

Job提交流程源码解析

e476693207ec468ea62ca60ca6ba622d.png

2、FileInputFormat切片源码解析


image.png

1.1.3 FileInputFormat 切片机制

1、切片机制

(1)简单地按照文件的内容长度进行切片

(2)切片大小。默认等于Block大小

(3)切片时不考虑数据集整体,而是逐个针对每一个文件单独切片

2、案例分析


image.png

3、源码解析

(1)源码中计算切片大小的公式

Math.max(minSize, Math.min(maxSize, blockSize));
mapreduce.input.fileinputformat.split.minsize=1 默认值为1
mapreduce.input.fileinputformat.split.maxsize= Long.MAXValue 默认值Long.MAXValue
因此,默认情况下,切片大小=blocksize。

(2)切片大小设置

maxsize(切片最大值):参数如果调得比blockSize小,则会让切片变小,而且就等于配置的这个参数的值。

minsize(切片最小值):参数调的比blockSize大,则可以让切片变得比blockSize还大。

(3)获取切片信息API

// 获取切片的文件名称
String name = inputSplit.getPath().getName();
// 根据文件类型获取切片信息
FileSplit inputSplit = (FileSplit) context.getInputSplit();

1.1.4 TextInputFormat

1、FileInputFormat实现类

FileInputFormat 常见的接口实现类包括:TextInputFormat、KeyValueTextInputFormat、

NLineInputFormat、CombineTextInputFormat 和自定义 InputFormat 等。

2、TextInputFormat

TextInputFormat 是默认的 FileInputFormat 实现类。按行读取每条记录。键是存储该行在整个文件中的起始字节偏移量, LongWritable 类型。值是这行的内容,不包括任何行终止符(换行符和回车符),Text 类型。

1.1.5 CombineTextInputFormat 切片机制

框架默认的TextInputFormat切片机制时对任务按文件规划切片,不管文件多小,都会时一个单独的切片,都会交给一个MapTask,这样如果有大量小文件,就会产生大量的MapTask,处理效率极其低下。

1、应用场景:

CombineTextInputFormat 用于小文件过多的场景,它可以将多个小文件从逻辑上规划到

一个切片中,这样,多个小文件就可以交给一个 MapTask 处理。

2、虚拟储存切片最大值设置

CombineTextInputFormat.setMaxInputSplitSize(job,4194304);//4M

注意:虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。

3、切片机制

生成切片过程包括:虚拟储存过程和切片过程两部分。

e999698ea06f4a46a61926020d990a9f.png

(1)虚拟储存过程:

将输入目录下所以文件大小,依次和设置的setMaxInputSpiltSize值比较,如果不大于设置的最大值,逻辑上划分一个块。如果输入文件大于设置·的最大值且大于两倍,那么以最大值切割一块;当剩余数据大小超过设置的最大值且不大于最大值两倍,此时将文件均分成2个虚拟储存块(防止出现太小切片)。

例如setMaxInputSpiltSize值为4M,输入文件大小为8.02M,先逻辑上分成一个4M。剩余的大小为4.02M,如果按照4M逻辑划分,就会出现0.02M的小的虚拟储存文件,所以将剩余的4.02文件切分成(2.01M,2.01M)两个文件。

(2)切片过程:

(a)判断虚拟储存的文件大小是否大于setMaxInputSpiltSize值,大于等于则单独形成一个切片。

(b)如果不大于则跟下一个虚拟储存文件进行合并,共同形成一个切片。

(c)测试举例: 有 4 个小文件大小分别为 1.7M、5.1M、3.4M 以及 6.8M 这四个小

文件,则虚拟存储之后形成 6 个文件块,大小分别为:1.7M,(2.55M、2.55M),3.4M 以及(3.4M、3.4M)

最终会形成三个切片,大小分别为:(1.7+2.55)M,(2.55+3.4)M,(3.4+3.4)M

1.1.6 CombineTextInputFormat案例实操

1、需求

将输入的大量小文件合并成一个切片统一处理

(1)输入数据

84bca849f9c84a0c8e1d0ec3c7b8dd9f.png

(2)期望

期望一个切片处理四个文件

2、实现过程

(1)正常情况下,这是要生成4个切片的

(2)在 WordcountDriver 中增加如下代码,运行程序,并观察运行的切片个数为 1。

// 如果不设置 InputFormat,它默认用的是 TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);
//虚拟存储切片最大值设置 20m   
CombineTextInputFormat.setMaxInputSplitSize(job, 20971520);

运行如果为 1 个切片:

number of splits:1

说明:为什么设置20M就是只有一个切片呢?

因为四个文件加起来都没有20M大(我们设置的MaxInputSpiltSize的大小)

1.2 MapReduce 工作流程

(1)

681f397b19294785998f8674f1fd379c.png

(2)

bbf5fe8c9a2a40878673e92ba1653485.png

上面的流程是整个MapReduce最全工作流程,但是Shuffle过程只是从第7步开始到第16步结束,具体Shuffle过程详解,如下:

(1)MapTask收集我们的map()方法输出的kv对,放到内存缓冲区中

(2)从内存缓冲区不断溢出到本地磁盘问就,可能会溢出多个文件

(3)多个溢出文件会被合并成大的溢出文件

(4)在溢出过程及合并过程中,都有调用Partitioner进行分区和针对key进行排序

(5)ReduceTask根据自己的分区号,去各个MapTask机器上取相对应的结果分区数据

(6)ReduceTask会抓取同一个分区的来着不同MapTask的结果文件,ReduceTask会将这些文件再进行合并(归并排序)

(7)合并成大文件后,Shuffle的过程也就结束了,后面进入ReduceTask的逻辑运算过程(从文件中取出一个一个的键值对Group,调用用户自定义的reduce()方法)

注意:

(1)Shuffle中的缓冲区大小会影响到MapReduce程序的执行效率,从原则上·说,缓冲器越大,磁盘IO的次数越少,执行的速度越快。

(2)缓冲区的大小可以通过参数调整,参数:mapreduce.task.io.sort.mb默认是100M


相关文章
|
4月前
|
分布式计算 Hadoop
Hadoop系列 mapreduce 原理分析
Hadoop系列 mapreduce 原理分析
40 1
|
3月前
|
存储 分布式计算 监控
Hadoop的JobTracker和TaskTracker在MapReduce中的作用是什么?
Hadoop的JobTracker和TaskTracker在MapReduce中的作用是什么?
56 0
|
4月前
|
存储 SQL 分布式计算
Hadoop(HDFS+MapReduce+Hive+数仓基础概念)学习笔记(自用)
Hadoop(HDFS+MapReduce+Hive+数仓基础概念)学习笔记(自用)
266 0
|
4月前
|
存储 分布式计算 资源调度
干翻Hadoop系列文章【03】:MapReduce概念详解
干翻Hadoop系列文章【03】:MapReduce概念详解
|
16天前
|
存储 分布式计算 Hadoop
大数据处理架构Hadoop
【4月更文挑战第10天】Hadoop是开源的分布式计算框架,核心包括MapReduce和HDFS,用于海量数据的存储和计算。具备高可靠性、高扩展性、高效率和低成本优势,但存在低延迟访问、小文件存储和多用户写入等问题。运行模式有单机、伪分布式和分布式。NameNode管理文件系统,DataNode存储数据并处理请求。Hadoop为大数据处理提供高效可靠的解决方案。
38 2
|
16天前
|
分布式计算 Hadoop 大数据
大数据技术与Python:结合Spark和Hadoop进行分布式计算
【4月更文挑战第12天】本文介绍了大数据技术及其4V特性,阐述了Hadoop和Spark在大数据处理中的作用。Hadoop提供分布式文件系统和MapReduce,Spark则为内存计算提供快速处理能力。通过Python结合Spark和Hadoop,可在分布式环境中进行数据处理和分析。文章详细讲解了如何配置Python环境、安装Spark和Hadoop,以及使用Python编写和提交代码到集群进行计算。掌握这些技能有助于应对大数据挑战。
|
18天前
|
SQL 分布式计算 Hadoop
利用Hive与Hadoop构建大数据仓库:从零到一
【4月更文挑战第7天】本文介绍了如何使用Apache Hive与Hadoop构建大数据仓库。Hadoop的HDFS和YARN提供分布式存储和资源管理,而Hive作为基于Hadoop的数据仓库系统,通过HiveQL简化大数据查询。构建过程包括设置Hadoop集群、安装配置Hive、数据导入与管理、查询分析以及ETL与调度。大数据仓库的应用场景包括海量数据存储、离线分析、数据服务化和数据湖构建,为企业决策和创新提供支持。
61 1
|
1月前
|
消息中间件 SQL 分布式计算
大数据Hadoop生态圈体系视频课程
熟悉大数据概念,明确大数据职位都有哪些;熟悉Hadoop生态系统都有哪些组件;学习Hadoop生态环境架构,了解分布式集群优势;动手操作Hbase的例子,成功部署伪分布式集群;动手Hadoop安装和配置部署;动手实操Hive例子实现;动手实现GPS项目的操作;动手实现Kafka消息队列例子等
20 1
大数据Hadoop生态圈体系视频课程
|
4月前
|
分布式计算 资源调度 搜索推荐
《PySpark大数据分析实战》-02.了解Hadoop
大家好!今天为大家分享的是《PySpark大数据分析实战》第1章第2节的内容:了解Hadoop。
48 0
《PySpark大数据分析实战》-02.了解Hadoop
|
4月前
|
存储 搜索推荐 算法
【大数据毕设】基于Hadoop的音乐推荐系统的设计和实现(六)
【大数据毕设】基于Hadoop的音乐推荐系统的设计和实现(六)
172 0