MapReduce框架--InputFormat数据输入--切片优化(11)

简介: MapReduce框架--InputFormat数据输入--切片优化(11)

MapReduce框架原理


这里的原理比较绕,搞了好久。还有点蒙。现在梳理下,防止忘记。


1.MapReduce工作流程

1dc618a0ed9580ce8bfa6facb208c08f.png5d4c6812c8535adbb050f4ddf2e1bce8.png


2)流程详解


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


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


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


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


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


5)reducetask根据自己的分区号,去各个maptask机器上取相应的结果分区数据


6)reducetask会取到同一个分区的来自不同maptask的结果文件,reducetask会将这些文件再进行合并(归并排序)


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


3)注意


Shuffle中的缓冲区大小会影响到mapreduce程序的执行效率,原则上说,缓冲区越大,磁盘io的次数越少,执行速度就越快。


缓冲区的大小可以通过参数调整,参数:io.sort.mb 默认100M。


2.InputFormat数据输入


1)Job提交流程和切片源码详解


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());


FileInputFormat源码解析(input.getSplits(job))


(1)找到你数据存储的目录。


(2)开始遍历处理(规划切片)目录下的每一个文件


(3)遍历第一个文件ss.txt


a)获取文件大小fs.sizeOf(ss.txt);


b)计算切片大小


Math.max(minSize, Math.min(maxSize, blockSize))

computeSliteSize(Math.max(minSize,Math.min(maxSize,blocksize)))=blocksize=128M


c)默认情况下,切片大小=blocksize


d)开始切,形成第1个切片:ss.txt—0:128M 第2个切片ss.txt—128:256M 第3个切片ss.txt—256M:300M(每次切片时,都要判断切完剩下的部分是否大于块的1.1倍,不大于1.1倍就划分一块切片)


e)将切片信息写到一个切片规划文件中


f)整个切片的核心过程在getSplit()方法中完成。


g)数据切片只是在逻辑上对输入数据进行分片,并不会再磁盘上将其切分成分片进行存储。InputSplit只记录了分片的元数据信息,比如起始位置、长度以及所在的节点列表等。


h)注意:block是HDFS物理上存储的数据,切片是对数据逻辑上的划分。


(4)提交切片规划文件到yarn上,yarn上的MrAppMaster就可以根据切片规划文件计算开启maptask个数。


2)FileInputFormat切片机制


FileInputFormat中默认的切片机制:


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


(2)切片大小,默认等于block大小


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


比如待处理数据有两个文件:


file1.txt 320M

file2.txt 10M


经过FileInputFormat的切片机制运算后,形成的切片信息如下:


file1.txt.split1-- 0~128

file1.txt.split2-- 128~256

file1.txt.split3-- 256~320

file2.txt.split1-- 0~10M


FileInputFormat切片大小的参数配置


通过分析源码,在FileInputFormat的280行中,计算切片大小的逻辑:Math.max(minSize, Math.min(maxSize, blockSize));


切片主要由这几个值来运算决定


mapreduce.input.fileinputformat.split.minsize=1 默认值为1

mapreduce.input.fileinputformat.split.maxsize= Long.MAXValue 默认值Long.MAXValue


因此,默认情况下,切片大小=blocksize。

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

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


获取切片信息API


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


3)CombineTextInputFormat切片机制


关于大量小文件的优化策略


1)默认情况下TextInputformat对任务的切片机制是按文件规划切片,不管文件多小,都会是一个单独的切片,都会交给一个maptask,这样如果有大量小文件,就会产生大量的maptask,处理效率极其低下。


2)优化策略


(1)最好的办法,在数据处理系统的最前端(预处理/采集),将小文件先合并成大文件,再上传到HDFS做后续分析。


(2)补救措施:如果已经是大量小文件在HDFS中了,可以使用另一种InputFormat来做切片(CombineTextInputFormat),它的切片逻辑跟TextFileInputFormat不同:它可以将多个小文件从逻辑上规划到一个切片中,这样,多个小文件就可以交给一个maptask。


(3)优先满足最小切片大小,不超过最大切片大小


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

CombineTextInputFormat.setMinInputSplitSize(job, 2097152);// 2m

举例:0.5m+1m+0.3m+5m=2m + 4.8m=2m + 4m + 0.8m


3)具体实现步骤


//  如果不设置InputFormat,它默认用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class)
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m
CombineTextInputFormat.setMinInputSplitSize(job, 2097152);// 2m


注:在看number of splits时,和最大值(MaxSplitSize)有关、总体规律就是和低于最大值是一片、高于最大值1.5倍+,则为两片;高于最大值2倍以上则向下取整,比如文件大小65MB,切片最大值为4MB,那么切片为16个.总体来说,切片差值不超过1个,不影响整体性能


4 InputFormat接口实现类


MapReduce任务的输入文件一般是存储在HDFS里面。输入的文件格式包括:基于行的日志文件、二进制格式文件等。这些文件一般会很大,达到数十GB,甚至更大。那么MapReduce是如何读取这些数据的呢?下面我们首先学习InputFormat接口。

InputFormat常见的接口实现类包括:TextInputFormat、KeyValueTextInputFormat、NLineInputFormat、CombineTextInputFormat和自定义InputFormat等。


1)TextInputFormat


TextInputFormat是默认的InputFormat。每条记录是一行输入。键K是LongWritable类型,存储该行在整个文件中的字节偏移量。值是这行的内容,不包括任何行终止符(换行符和回车符)。


以下是一个示例,比如,一个分片包含了如下4条文本记录。


Rich learning form
Intelligent learning engine
Learning more convenient
From the real demand for more close to the enterprise


每条记录表示为以下键/值对:


(0,Rich learning form)
(19,Intelligent learning engine)
(47,Learning more convenient)
(72,From the real demand for more close to the enterprise)


很明显,键并不是行号。一般情况下,很难取得行号,因为文件按字节而不是按行切分为分片。


2)KeyValueTextInputFormat


每一行均为一条记录,被分隔符分割为key,value。可以通过在驱动类中设置conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPERATOR, " ");来设定分隔符。默认分隔符是tab(\t)。


以下是一个示例,输入是一个包含4条记录的分片。其中——>表示一个(水平方向的)制表符。


line1 ——>Rich learning form
line2 ——>Intelligent learning engine
line3 ——>Learning more convenient
line4 ——>From the real demand for more close to the enterprise


此时的键是每行排在制表符之前的Text序列。


3)NLineInputFormat


如果使用NlineInputFormat,代表每个map进程处理的InputSplit不再按block块去划分,而是按NlineInputFormat指定的行数N来划分。即输入文件的总行数/N=切片数(20),如果不整除,切片数=商+1。


以下是一个示例,仍然以上面的4行输入为例。


Rich learning form
Intelligent learning engine
Learning more convenient
From the real demand for more close to the enterprise


例如,如果N是2,则每个输入分片包含两行。开启2个maptask。


(0,Rich learning form)
(19,Intelligent learning engine)


另一个 mapper 则收到后两行:


(47,Learning more convenient)
(72,From the real demand for more close to the enterprise)


这里的键和值与TextInputFormat生成的一样。


4)自定义InputFormat


1)概述


(1)自定义一个类继承FileInputFormat。


(2)改写RecordReader,实现一次读取一个完整文件封装为KV。


(3)在输出时使用SequenceFileOutPutFormat输出合并文件。


2)案例实操


详见7.4小文件处理(自定义InputFormat)。


相关文章
|
21天前
|
数据采集 分布式计算 Hadoop
使用Hadoop MapReduce进行大规模数据爬取
使用Hadoop MapReduce进行大规模数据爬取
|
2月前
|
SQL 分布式计算 关系型数据库
Hadoop-24 Sqoop迁移 MySQL到Hive 与 Hive到MySQL SQL生成数据 HDFS集群 Sqoop import jdbc ETL MapReduce
Hadoop-24 Sqoop迁移 MySQL到Hive 与 Hive到MySQL SQL生成数据 HDFS集群 Sqoop import jdbc ETL MapReduce
110 0
|
2月前
|
SQL 分布式计算 关系型数据库
Hadoop-23 Sqoop 数据MySQL到HDFS(部分) SQL生成数据 HDFS集群 Sqoop import jdbc ETL MapReduce
Hadoop-23 Sqoop 数据MySQL到HDFS(部分) SQL生成数据 HDFS集群 Sqoop import jdbc ETL MapReduce
51 0
|
2月前
|
SQL 分布式计算 关系型数据库
Hadoop-22 Sqoop 数据MySQL到HDFS(全量) SQL生成数据 HDFS集群 Sqoop import jdbc ETL MapReduce
Hadoop-22 Sqoop 数据MySQL到HDFS(全量) SQL生成数据 HDFS集群 Sqoop import jdbc ETL MapReduce
60 0
|
7月前
|
存储 分布式计算 监控
Hadoop【基础知识 01+02】【分布式文件系统HDFS设计原理+特点+存储原理】(部分图片来源于网络)【分布式计算框架MapReduce核心概念+编程模型+combiner&partitioner+词频统计案例解析与进阶+作业的生命周期】(图片来源于网络)
【4月更文挑战第3天】【分布式文件系统HDFS设计原理+特点+存储原理】(部分图片来源于网络)【分布式计算框架MapReduce核心概念+编程模型+combiner&partitioner+词频统计案例解析与进阶+作业的生命周期】(图片来源于网络)
339 2
|
3月前
|
分布式计算 资源调度 Hadoop
在YARN集群上运行部署MapReduce分布式计算框架
主要介绍了如何在YARN集群上配置和运行MapReduce分布式计算框架,包括准备数据、运行MapReduce任务、查看任务日志,并启动HistoryServer服务以便于日志查看。
77 0
|
4月前
|
缓存 分布式计算 Java
详细解读MapReduce框架中的分布式缓存
【8月更文挑战第31天】
59 0
|
4月前
|
缓存 分布式计算 算法
优化Hadoop MapReduce性能的最佳实践
【8月更文第28天】Hadoop MapReduce是一个用于处理大规模数据集的软件框架,适用于分布式计算环境。虽然MapReduce框架本身具有很好的可扩展性和容错性,但在某些情况下,任务执行可能会因为各种原因导致性能瓶颈。本文将探讨如何通过调整配置参数和优化算法逻辑来提高MapReduce任务的效率。
674 0
|
4月前
|
存储 分布式计算 分布式数据库
《HBase MapReduce之旅:我的学习笔记与心得》——跟随我的步伐,一同探索HBase世界,揭开MapReduce的神秘面纱,分享那些挑战与收获,让你在数据的海洋里畅游无阻!
【8月更文挑战第17天】HBase是Apache顶级项目,作为Bigtable的开源版,它是一个非关系型、分布式数据库,具备高可扩展性和性能。结合HDFS存储和MapReduce计算框架,以及Zookeeper协同服务,HBase支持海量数据高效管理。MapReduce通过将任务拆解并在集群上并行执行,极大提升处理速度。学习HBase MapReduce涉及理解其数据模型、编程模型及应用实践,虽然充满挑战,但收获颇丰,对职业发展大有裨益。
58 0
|
7月前
|
分布式计算 并行计算 搜索推荐
Hadoop MapReduce计算框架
【5月更文挑战第10天】HadoopMapReduce计算框架
63 3

热门文章

最新文章