码迷,mamicode.com
首页 > 其他好文 > 详细

hadoop 查询优化

时间:2021-06-21 21:16:06      阅读:0      评论:0      收藏:0      [点我收藏+]

标签:最大值   数据结构   下载   combine   存在   time   实现类   使用   机器   

小文件问题处理

  像hadoop这样的分布式集群是适合大数据文件处理的,为什么不适合小文件处理呢?

    1.存储:hadoop的存储每个文件都会在NameNode上记录元数据,如果同样大小的文件,文件很小的话,就会产生很多文件,造成NameNode的压力;

    2.读取:同样大小的文件分为很多小文件的话,会增加磁盘寻址次数,降低性能;

    3.计算:我们知道一个map默认处理一个分片或者一个小文件,如果map的启动时间都比数据处理的时间还要长,那么就会造成性能低,而且在map端溢写磁盘的时候每一个map最终会产生reduce数量个数的中间结果,如果map数量特别多,就会造成临时文件很多,而且在reduce拉取数据的时候增加磁盘的IO.

那如果在使用的过程中真的遇到小文件的情况怎么办呢?

    1.从源头干掉,也就是在hdfs上我们不存储小文件,也就是数据上传hdfs的时候我们就合并小文件;
    2.在FileInputFormat读取入数据的时候我们使用实现类CombineFileInputFormat读取数据,在读取数据的时候进行合并.

数据倾斜问题处理

    从hadoop运行机制上可以了解到,mapreduce是一个并行处理,那么处理的时间肯定是作业中所有任务最慢的那个了,也就是木桶效应

      1.数据倾斜:每个reduce处理的数据量不是同一个量级的,导致大部分任务已经执行完成,但是存在少数几个任务执行的很慢、很慢,导致整个作业的执行时间很长;
      2.还有可能就是某些作业所在的NodeManager有问题或者container有问题,导致作业执行缓慢。

    那么为什么会产生数据倾斜呢?

      1.数据本身就不平衡,所以在默认的hashpartition时造成分区数据不一致问题;

      2.代码设计缺陷或者设计不合理.

    那如何解决数据倾斜的问题呢?

      1.分区导致倾斜市可根据业务的特点,自定义合理的分区键,使得每个分区数据基本均匀
      2.不符合hash的,让其符合hash分区,如:在key前加随机数n-key。
      3.如果是reduce处理慢,我们可以增加reduce的内存和vcore呀,这样挺高性能就快了,虽然没从根本上解决问题,但是还有明显效果

 

mapreduce过程优化

    对于map阶段来说首先第一步可以优化处理的是map读取和map数大小,在本文开头有说到小文件的问题,所以在数据的读取这里也可以做优化,另外就是map数,从hadoop的运行原理可以了解到hadoop对数据文件是进行分片读取的,分片的计算公:Math.max(minSize, Math.min(maxSize, blockSize)),minSize的默认值是1,而maxSize的默认值是long类型的最大值,即可得切片的默认大小是blockSize(128M),maxSize参数如果调得比blocksize小,则会让切片变小,而且就等于配置的这个参数的值,minSize参数调的比blockSize大,则可以让切片变得比blocksize还大,由于map数没有具体的参数指定,所以我们可以通过如上的公式调整切片的大小,这样我们就可以设置map数了,那么问题来了,map数该如何设置呢?这些东西一定要结合业务,map数太多,会产生很多中间结果,导致reduce拉取数据变慢,太少,每个map处理的时间又很长,需要结合实际的业务需求,一般情况下可以把map的执行时间调至到一分钟左右比较合适,如果数据量就是很大,有时候还是需要控制map的数量,这个时候每个map的执行时间就比较长了,这种情况下我们还是可以做一定的优化的,比如通过调整map的处理能力,入调大map阶段的内存大小:mapreduce.map.memory.mb=3G(默认1G)mapreduce.map.cpu.vcores=1(默认也是1),这些就是map任务启动前可以进行的优化处理了,之后就进入map的写环形换冲区,为什么需要写环形换冲区,而不是直接写磁盘呢?这样的目的主要是为了减少磁盘I/O,每个Map任务不断地将键值对输出到在内存中构造的一个环形数据结构中。使用环形数据结构是为了更有效地使用内存空间,在内存中放置尽可能多的数据。执行流程是,该缓冲默认100M(mapreduce.task.io.sort.mb参数控制),当到达80%(mapreduce.map.sort.spill.percent参数控制)时就会溢写磁盘。每达到80%都会重写溢写到一个新的文件。那么完全可以根据机器的配置和数据来两种这两个参数,当内存足够,增大mapreduce.task.io.sort.mb完全会提高溢写的过程,而且会减少中间结果的文件数量。比如调整mapreduce.task.io.sort.mb=512。当文件溢写完后,会对这些文件进行合并,默认每次合并10(mapreduce.task.io.sort.factor参数控制)个溢写的文件,调整mapreduce.task.io.sort.factor=64。这样可以提高合并的并行度,减少合并的次数,降低对磁盘操作的次数。

    之后就到了map到reduce的数据传输过程了:

      这中间主要的影响无非就是磁盘IO,网络IO,数据量的大小了(是否压缩),其实减少数据量的大小,就可以做到优化了,所以我们可以选择性压缩数据,这样在传输的过程中
就可以降低磁盘IO,网络IO等。可以通过mapreduce.map.output.compress(default:false)设置为true进行压缩,数据会被压缩写入磁盘,读数据读的是压缩数据需要解压,在实际经验中Hive在Hadoop的运行的瓶颈一般都是IO而不是CPU,压缩一般可以10倍的减少IO操作,压缩的方式Gzip,Lzo,BZip2,Lzma等,其中Lzo是一种比较平衡选择,mapreduce.map.output.compress.codec(default:org.apache.hadoop.io.compress.DefaultCodec)参数设置。

mapreduce.task.io.sort.mb        #排序map输出所需要使用内存缓冲的大小,以兆为单位, 默认为100
mapreduce.map.sort.spill.percent #map输出缓冲和用来磁盘溢写过程的记录边界索引,这两者使用的阈值,默认0.8
mapreduce.task.io.sort.factor    #排序文件时,一次最多合并的文件数,默认10
mapreduce.map.output.compress    #在map溢写磁盘的过程是否使用压缩,默认false
org.apache.hadoop.io.compress.SnappyCodec  #map溢写磁盘的压缩算法,默认org.apache.hadoop.io.compress.DefaultCodec
mapreduce.shuffle.max.threads    #该参数表示每个节点管理器的工作线程,用于map输出到reduce,默认为0,表示可用处理器的两倍

    接下来就是reduce了,首先可以通过参数设置合理的reduce个数(mapreduce.job.reduces参数控制),以及通过参数设置每个reduce的资源,mapreduce.reduce.memory.mb=5G(默认1G)mapreduce.reduce.cpu.vcores=1(默认为1)。

reduce在copy的过程中默认使用5(mapreduce.reduce.shuffle.parallelcopies参数控制)个并行度进行复制数据,mapreduce.reduce.shuffle.parallelcopies=100.reduce的每一个下载线程在下载某个map数据的时候,有可能因为那个map中间结果所在机器发生错误,或者中间结果的文件丢失,或者网络瞬断等等情况,这样reduce的下载就有可能失败,所以reduce的下载线程并不会无休止的等待下去,当一定时间后下载仍然失败,那么下载线程就会放弃这次下载,并在随后尝试从另外的地方下载(因为这段时间map可能重跑)。reduce下载线程的这个最大的下载时间段是可以通过mapreduce.reduce.shuffle.read.timeout(default180000秒)调整的。

Copy过来的数据会先放入内存缓冲区中,然后当使用内存达到一定量的时候才spill磁盘。这里的缓冲区大小要比map端的更为灵活,它基于JVM的heap size设置。这个内存大小的控制就不像map一样可以通过io.sort.mb来设定了,而是通过另外一个参数 mapreduce.reduce.shuffle.input.buffer.percent(default 0.7)控制的。意思是说,shuffile在reduce内存中的数据最多使用内存量为:0.7 × maxHeap of reduce task,内存到磁盘merge的启动门限可以通过mapreduce.reduce.shuffle.merge.percent(default0.66)配置。

copy完成后,reduce进入归并排序阶段,合并因子默认为10(mapreduce.task.io.sort.factor参数控制),如果map输出很多,则需要合并很多趟,所以可以提高此参数来减少合并次数。

mapreduce.reduce.shuffle.parallelcopies #把map输出复制到reduce的线程数,默认5
mapreduce.task.io.sort.factor  #排序文件时一次最多合并文件的个数
mapreduce.reduce.shuffle.input.buffer.percent #在shuffle的复制阶段,分配给map输出缓冲区占堆内存的百分比,默认0.7
mapreduce.reduce.shuffle.merge.percent #map输出缓冲区的阈值,用于启动合并输出和磁盘溢写的过程



部分借鉴https://www.cnblogs.com/zsql/p/11614580.html

hadoop 查询优化

标签:最大值   数据结构   下载   combine   存在   time   实现类   使用   机器   

原文地址:https://www.cnblogs.com/zzuyczhang/p/14915053.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!