0


hive/spark数据倾斜解决方案

在这里插入图片描述

Hive数据倾斜以及解决方案

1、什么是数据倾斜

数据倾斜主要表现在,mapreduce程序执行时,reduce节点大部分执行完毕,但是有一个或者几个reduce节点运行很慢,导致整个程序的处理时间很长,这是因为某一个key的条数比其他key多很多(有时是百倍或者千倍之多),这条Key所在的reduce节点所处理的数据量比其他节点就大很多,从而导致某几个节点迟迟运行不完。
在这里插入图片描述

2、数据倾斜的原因及现象

一些操作导致的数据倾斜:
在这里插入图片描述
主要原因:

  • key分布不均匀
  • 业务数据本身的特性
  • 建表时考虑不周
  • 某些SQL语句本身就有数据倾斜

现象:
任务进度长时间维持在99%(或100%),查看任务监控页面,发现只有少量(1个或几个)reduce子任务未完成。因为其处理的数据量和其他reduce差异过大。
单一reduce的记录数与平均记录数差异过大,通常可能达到3倍甚至更多。 最长时长远大于平均时长。

3、数据倾斜的解决方案

1)参数调节

  • Map 端部分聚合,相当于Combiner
hive.map.aggr =true
  • 有数据倾斜的时候进行负载均衡,当选项设定为true,生成的查询计划会有两个MR Job。 第一个MR Job中,Map 的输出结果集合会随机分布到Reduce中,每个Reduce做部分聚合操作,并输出结果,这样处理 的结果是相同的Group By Key有可能被分发到不同的Reduce中,从而达到负载均衡的目的; 第二个MR Job再根据预处理的数据结果按照Group By Key分布到Reduce中(这个过程可以保证相同的Group By Key 被分布到同一个Reduce中),最后完成最终的聚合操作。
hive.groupby.skewindata=true

2)SQL语句调节

如何join:
关于驱动表的选取,选用join key分布最均匀的表作为驱动表,做好列裁剪和filter操作,以达到两表做
join的时候,数据量相对变小的效果。

大小表Join:
使用map join让小的维度表(1000条以下的记录条数)先进内存。在map端完成reduce。

大表Join大表:
把空值的key变成一个字符串加上随机数,把倾斜的数据分到不同的reduce上,由于null值关联不上,处理后并不影响最终结果。

count distinct大量相同特殊值:
count distinct时,将值为空的情况单独处理,如果是计算count distinct,可以不用处理,直接过滤,在
最后结果中加1。如果还有其他计算,需要进行group by,可以先将值为空的记录单独处理,再和其他计
算结果进行union。

group by维度过小:
采用sum() group by的方式来替换count(distinct)完成计算。

特殊情况特殊处理:
在业务逻辑优化效果的不大情况下,有些时候是可以将倾斜的数据单独拿出来处理。最后union回去。

4、典型的业务场景

1)空值产生的数据倾斜

场景:如日志中,常会有信息丢失的问题,比如日志中的 user_id,如果取其中的 user_id 和 用户表中的
user_id 关联,会碰到数据倾斜的问题。

解决方法一:user_id为空的不参与关联

select * from log a
joinusers b
on a.user_id is not null
and a.user_id = b.user_id
union all
select * from log a
where a.user_id is null;

解决方法二:赋与空值分新的key值

select *
from log a
left outer joinusers b
on case when a.user_id is null then concat('hive',rand())else a.user_id end =
b.user_id;

结论:
方法二比方法1、一效率更好,不但io少了,而且作业数也少了。
解决方法一中log读取两次,jobs是2。
解决方法二job数是1。
这个优化适合无效id (比如 -99 , ‘’, null 等) 产生的倾斜问题。把空值的key变成一个字符串加上随机数,就能把倾斜的数据分到不同的reduce上,解决数据倾斜问题。

2)不同数据类型关联产生数据倾斜

场景:用户表中user_id字段为int,log表中user_id字段既有string类型也有int类型。
当按照user_id进行两个表的Join操作时,默认的Hash操作会按int型的id来进行分配,这样会导致所有string类型id的记录都分
配到一个Reducer中。

解决方法:把数字类型转换成字符串类型

select * from users a
left outer join logs b
on a.usr_id = cast(b.user_id as string);

3)小表不小不大,怎么用 map join 解决倾斜问题

使用map join解决小表(记录数少)关联大表的数据倾斜问题,这个方法使用的频率非常高,但如果小表很大,大到map join会出现bug或异常,这时就需要特别的处理。
例子:

select * from log a
left outer joinusers b
on a.user_id = b.user_id;

users表有600w+的记录,把users分发到所有的map上也是个不小的开销,而且map join不支持这么大的小表。如果用普通的join,又会碰到数据倾斜的问题

解决方法:

select /*+mapjoin(x)*/* from log a
left outer join(select /*+mapjoin(c)*/d.*
from (select distinct user_id from log ) c
joinusers d
on c.user_id = d.user_id
) x
on a.user_id = b.user_id;

假如,log里user_id有上百万个,这就又回到原来map join问题。所幸,每日的会员uv不会太多,有交易
的会员不会太多,有点击的会员不会太多,有佣金的会员不会太多等等。所以这个方法能解决很多场景
下的数据倾斜问题。

Hive如果不用参数调优,在map和reduce端应该做什么

1、map阶段优化

Map阶段的优化,主要是确定合适的map数。那么首先要了解map数的计算公式

num_reduce_tasks = min[${hive.exec.reducers.max},(${input.size}/${hive.exec.reducers.bytes.per.reducer})]
  • mapred.min.split.size: 指的是数据的最小分割单元大小;min的默认值是1B
  • mapred.max.split.size: 指的是数据的最大分割单元大小;max的默认值是256MB
  • dfs.block.size: 指的是HDFS设置的数据块大小。个已经指定好的值,而且这个参数默认情况下hive是识别不到的。

通过调整max可以起到调整map数的作用,减小max可以增加map数,增大max可以减少map数。需要提醒的是,直接调整mapred.map.tasks这个参数是没有效果的。

2、reduce阶段优化

这里说的reduce阶段,是指前面流程图中的reduce phase(实际的reduce计算)而非图中整个reduce
task。Reduce阶段优化的主要工作也是选择合适的reduce task数量, 与map优化不同的是,reduce优化时,可以直接设置mapred.reduce.tasks参数从而直接指定reduce的个数。

num_reduce_tasks = min[${hive.exec.reducers.max},(${input.size}/${hive.exec.reducers.bytes.per.reducer})]

hive.exec.reducers.max :此参数从Hive 0.2.0开始引入。在Hive 0.14.0版本之前默认值是999;而从
Hive 0.14.0开始,默认值变成了1009,这个参数的含义是最多启动的Reduce个数

hive.exec.reducers.bytes.per.reducer :此参数从Hive 0.2.0开始引入。在Hive 0.14.0版本之前默
认值是1G(1,000,000,000);而从Hive 0.14.0开始,默认值变成了256M(256,000,000),可以参见HIVE-7158和
HIVE-7917。这个参数的含义是每个Reduce处理的字节数。比如输入文件的大小是1GB,那么会启动4个Reduce来处理数据。

也就是说,根据输入的数据量大小来决定Reduce的个数,默认Hive.exec.Reducers.bytes.per.Reducer为1G,而且Reduce个数不能超过一个上限参数值,这个参数的默认取值为999。所以我们可以调整Hive.exec.Reducers.bytes.per.Reducer来设置Reduce个数。

注意:
Reduce的个数对整个作业的运行性能有很大影响。
如果Reduce设置的过大,那么将会产生很多小文件,对NameNode会产生一定的影响,而且整个作业的运行时间未必会减少;
如果Reduce设置的过小,那么单个Reduce处理的数据将会加大,很可能会引起OOM异常。

如果设置了 mapred.reduce.tasks/mapreduce.job.reduces 参数,那么Hive会直接使用它的值作为Reduce的个数;
如果mapred.reduce.tasks/mapreduce.job.reduces的值没有设置(也就是-1),那么Hive会根据输入文件的大小估算出Reduce的个数。
根据输入文件估算Reduce的个数可能未必很准确,因为Reduce的输入是Map的输出,而Map的输出可能会比输入要小,所以最准确的数根据Map的输出估算Reduce的个数。

Spark数据倾斜问题,如何定位,解决方案

在这里插入图片描述

1、数据倾斜

Spark中的数据倾斜问题主要指shuffle过程中出现的数据倾斜问题,是由于不同的key对应的数据量不同
导致的不同task所处理的数据量不同的问题。
例如,reduce点一共要处理100万条数据,第一个和第二个task分别被分配到了1万条数据,计算5分钟内完成,第三个task分配到了98万数据,此时第三个task可能需要10个小时完成,这使得整个Spark作业需要10个小时才能运行完成,这就是数据倾斜所带来的后果

数据倾斜俩大直接致命后果
1)数据倾斜直接会导致一种情况:Out Of Memory
2)运行速度慢
注意,要区分开数据倾斜与数据量过量这两种情况
数据倾斜是指少数task被分配了绝大多数的数据,因此少数task运行缓慢;
数据过量是指所有task被分配的数据量都很大,相差不多,所有task都运行缓慢

数据倾斜的表现:
1)Spark作业的大部分task都执行迅速,只有有限的几个task执行的非常慢,此时可能出现了数据倾斜,作业可以运行,但是运行得非常慢,严重影响出数时间
2)Spark作业的大部分task都执行迅速,但是有的task在运行过程中会突然报出OOM,反复执行几次都在某一个task报出OOM错误,此时可能出现了数据倾斜,作业无法正常运行

定位数据倾斜问题:
1)查阅代码中的shuffe算子,例如reduceByKey、countByKey、groupByKey、join等算子,根据代码逻辑判断此处是否会出现数据倾斜
2)查看Spark作业的log文件,log文件对于错误的记录会精确到代码的某一行,可以根据异常定位到的代码位置来明确错误发生在第几个stage,对应的shuffle算子是哪一个

2、解决方案一:聚合元数据

1)避免shuffle过程

绝大多数情况下,Spark作业的数据来源都是Hive表,这些Hive表基本都是经过ETL之后的昨天的数据为了避免数据倾斜,我们可以考虑避免shuffle过程,如果避免了shuffle过程,那么从根本上就消除了发生数据倾斜问题的可能

如果Spark作业的数据来源于Hive表,那么可以先在Hive表中对数据进行聚合,例如按照key进行分组,将同一key对应的所有value用一种特殊的格式拼接到一个字符串里去,这样,一个key就只有一条数据了;
之后,对一个key的所有value进行处理时,只需要进行map操作即可,无需再进行任何的shuffle操作。

通过上述方式就避免了执行shuffle操作,也就不可能会发生任何的数据倾斜问题。

对于Hive表中数据的操作,不一定是拼接成一个字符串,也可以是直接对key的每一条数据进行累计计算要区分开,处理的数据量大和数据倾斜的区别

2)缩小key粒度(增大数据倾斜可能性,降低每个task的数据量)

key的数量增加,可能使数据倾斜更严重

3)增大key粒度(减小数据倾斜可能性,增大每个task的数据量)

如果没有办法对每个key聚合出来一条数据,在特定场景下,可以考虑扩大key的聚合粒度

例如,目前有10万条用户数据,当前key的粒度是(省,城市,区,日期),现在我们考虑扩大粒度,将key的粒度扩大为(省,城市,日期),这样的话,key的数量会减少,key之间的数据量差异也有可能会减少,由此可以减轻数据倾斜的现象和问题。(此方法只针对特定类型的数据有效,当应用场景不适宜时,会加重数据倾斜)

3、解决方案二:过滤导致倾斜的key

如果在Spark作业中允许丢弃某些数据,那么可以考虑将可能导致数据倾斜的key进行过滤,滤除可能导致数据倾斜的key对应的数据,这样,在Spark作业中就不会发生数据倾斜了

4、解决方案三:提高shuffle操作中的reduce并行度

当方案一和方案二对于数据倾斜的处理没有很好的效果时,可以考虑提高shuffle过程中的reduce端并行度,reduce端并行度的提高就增加了reduce端task的数量,那么每个task分配到的数据量就会相应减少,由此缓解数据倾斜问题

1)reduce端并行度的设置

在大部分的shuffle算子中,都可以传入一个并行度的设置参数,比如reduceByKey(500),这个参数会决定shuffle过程中reduce端的并行度,在进行shuffle操作的时候,就会对应着创建指定数量的reduce task。对于Spark SQL中的shuffle类语句,比如group by、join等,需要设置一个参数,即spark.sql.shuffle.partitions,该参数代表了shuffle read task的并行度,该值默认是200,对于很多场景来说都有点过小

增加shuffle read task的数量,可以让原本分配给一个task的多个key分配给多个task,从而让每个task处理比原来更少的数据。
举例来说,如果原本有5个key,每个key对应10条数据,这5个key都是分配给一个task的,那么这个task就要处理50条数据。
而增加了shuffle read task以后,每个task就分配到一个key,即每个task就处理10条数据,那么自然每个task的执行时间都会变短了

2)reduce端并行度设置存在的缺陷

提高reduce端并行度并没有从根本上改变数据倾斜的本质和问题(方案一和方案二从根本上避免了数据倾斜的发生),只是尽可能地去缓解和减轻shuffle reduce task的数据压力,以及数据倾斜的问题,适用于有较多key对应的数据量都比较大的情况
该方案通常无法彻底解决数据倾斜,因为如果出现一些极端情况,比如某个key对应的数据量有100万,那么无论你的task数量增加到多少,这个对应着100万数据的key肯定还是会分配到一个task中去处理,因此注定还是会发生数据倾斜的。
所以这种方案只能说是在发现数据倾斜时尝试使用的第一种手段,尝试去用最简单的方法缓解数据倾斜而已,或者是和其他方案结合起来使用
在理想情况下,reduce端并行度提升后,会在一定程度上减轻数据倾斜的问题,甚至基本消除数据倾斜;
但是,在一些情况下,只会让原来由于数据倾斜而运行缓慢的task运行速度稍有提升,或者避免了某些task的OOM问题,但是,仍然运行缓慢,此时,要及时放弃方案三,开始尝试后面的方案

5、解决方案四:使用随机key实现双重聚合

当使用了类似于groupByKey、reduceByKey这样的算子时,可以考虑使用随机key实现双重聚合,如下图所示
在这里插入图片描述
首先,通过map算子给每个数据的key添加随机数前缀,对key进行打散,将原先一样的key变成不一样的key;
然后进行第一次聚合,这样就可以让原本被一个task处理的数据分散到多个task上去做局部聚合;
随后,去除掉每个key的前缀,再次进行聚合此方法对于由groupByKey、reduceByKey这类算子造成的数据倾斜由比较好的效果,仅仅适用于聚合类的shuffle操作,适用范围相对较窄。如果是join类的shuffle操作,还得用其他的解决方案。

此方法也是前几种方案没有比较好的效果时要尝试的解决方案

6、解决方案五:将reduce join转换为map join

正常情况下,join操作都会执行shuffle过程,并且执行的是reduce join,也就是先将所有相同的key和对应的value汇聚到一个reduce task中,然后再进行join。普通join的过程如下图所示
在这里插入图片描述

普通的join是会走shuffle过程的,而一旦shuffle,就相当于会将相同key的数据拉取到一个shuffle read task中再进行join,此时就是reduce join。但是如果一个RDD是比较小的,则可以采用广播小RDD全量数据+map算子来实现与join同样的效果,也就是map join,此时就不会发生shuffle操作,也就不会发生数据倾斜

注意,RDD是并不能进行广播的,只能将RDD内部的数据通过collect拉取到Driver内存然后再进行广播

1)核心思路

不使用join算子进行连接操作,而使用Broadcast变量与map类算子实现join操作,进而完全规避掉shuffle类的操作,彻底避免数据倾斜的发生和出现。

将较小RDD中的数据直接通过collect算子拉取到Driver端的内存中来,然后对其创建一个Broadcast变量;
接着对另外一个RDD执行map类算子,在算子函数内,从Broadcast变量中获取较小RDD的全量数据,与当前RDD的每一条数据按照连接key进行比对,如果连接key相同的话,那么就将两个RDD的数据用你需要的方式连接起来
根据上述思路,根本不会发生shuffle操作,从根本上杜绝了join操作可能导致的数据倾斜问题。

当join操作有数据倾斜问题并且其中一个RDD的数据量较小时,可以优先考虑这种方式,效果非常好。
map join的过程如下图所示
在这里插入图片描述

2)不使用场景分析

由于Spark的广播变量是在每个Executor中保存一个副本,如果两个RDD数据量都比较大,那么如果将一个数据量比较大的RDD做成广播变量,那么很有可能会造成内存溢出

7、解决方案六:sample采样对倾斜key单独进行join

在Spark中,如果某个RDD只有一个key,那么在shuffle过程中会默认将此key对应的数据打散,由不同的
reduce端task进行处理

当由单个key导致数据倾斜时,可有将发生数据倾斜的key单独提取出来,组成一个RDD,然后用这个原本会导致倾斜的key组成的RDD根其他RDD单独join,此时,根据Spark的运行机制,此RDD中的数据会在shuffle阶段被分散到多个task中去进行join操作。倾斜key单独join的流程如下图所示
在这里插入图片描述
1)适用场景分析
对于RDD中的数据,可以将其转换为一个中间表,或者是直接使用countByKey()的方式,看一个这个RDD中各个key对应的数据量,此时如果你发现整个RDD就一个key的数据量特别多,那么就可以考虑使用这种方法

当数据量非常大时,可以考虑使用sample采样获取10%的数据,然后分析这10%的数据中哪个key可能会导致数据倾斜,然后将这个key对应的数据单独提取出来

Map Join能解决数据倾斜的原因

Map Join概念:将其中做连接的小表(全量数据)分发到所有 MapTask 端进行 Join,从而避免了reduceTask,前提要求是内存足以装下该全量数据。

Map Join通常用于一个很小的表和一个大表进行join的场景,具体小表有多小,由参数hive.mapjoin.smalltable.filesize来决定,该参数表示小表的总大小,默认值为25000000字节,即25M。 一般默认就够了,无须修改。

1)在多表关联情况下,将小表(关联键记录少的表)依次放到前面,这样能够触发reduce端减少操作次数,从而减少运行时间。

2)同时使用Map Join让小表缓存到内存。在map端完成join过程,这样就能省掉redcue端的工作。

需要注意:这一功能使用时,需要开启map-side join的设置属性:

set hive.auto.convert.join=true(默认是false)

注意: 大表放硬盘,小表放内存( 前提要求是内存足以装下该全量数据 )。

标签: hive spark 大数据

本文转载自: https://blog.csdn.net/weixin_37791303/article/details/131302487
版权归原作者 临风赏月 所有, 如有侵权,请联系我们删除。

“hive/spark数据倾斜解决方案”的评论:

还没有评论