Hive SQL使用Skewed方式建表解决数据倾斜小记
Posted 扫地增
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Hive SQL使用Skewed方式建表解决数据倾斜小记相关的知识,希望对你有一定的参考价值。
Skewed Table
可以提高有一个或多个列有倾斜值的表的性能,通过指定经常出现的值(严重倾斜),hive
将会在元数据中记录这些倾斜的列名和值,在join时能够进行优化。若是指定了STORED AS DIRECTORIES
,也就是使用列表桶(ListBucketing)
,hive
会对倾斜的值建立子目录,查询会更加得到优化。
创建表是指定为 Skewed Table
,如下例子,STORED AS DIRECTORIES
是可选择的,它指定了使用列表桶(ListBucketing)
。
单倾斜列建表
CREATE TABLE list_bucket_single
(
key STRING,
value STRING
)
SKEWED BY (key) ON (1,5,6)
[STORED AS DIRECTORIES];
你也可以使用多个列创建Skewed Table
,如下使用两列。
多倾斜列建表
CREATE TABLE list_bucket_multiple
(
col1 STRING,
col2 int,
col3 STRING
)
SKEWED BY (col1, col2) ON (('s1',1), ('s3',3), ('s13',13), ('s78',78))
[STORED AS DIRECTORIES];
使用alter table修改表倾斜信息
可以使用alter table
语句来对已创建的表修改倾斜信息。
ALTER TABLE <T> (SCHEMA) SKEWED BY (keys) ON ('c1', 'c2') [STORED AS DIRECTORIES];
这个语句仅支持表级别,不支持分区级别。它可以将一个普通表转为倾斜表,也可以修改倾斜表的倾斜列名称和值。它会影响到之后创建的分区,对
alter
语句之前的分区无影响。
关闭倾斜特征
ALTER TABLE <T> (SCHEMA) NOT SKEWED;
关闭倾斜特征,使一个表成为非倾斜表,同样会影响到之后创建的分区,对
alter
语句之前的分区无影响。
关闭list bucketing
关闭list bucketing
,表成为普通的倾斜表。
ALTER TABLE <T> (SCHEMA) NOT STORED AS DIRECTORIES;
修改list bucketing倾斜值的存储位置映射
修改list bucketing倾斜值的存储位置映射。
ALTER TABLE <T> (SCHEMA) SET SKEWED LOCATION (key1="loc1", key2="loc2");
Skewed Join Optimization
两个大数据表的连接由一组
MapReduce作
业完成,它们首先根据连接键对表进行排序,然后将它们连接起来,mapper
将相同键的行发送到同一个reduce
。
假设表A id
字段有值1,2,3,4
,并且表B
也含有id
列,含有值1,2,3
。我们使用如下语句来进行连接。
select
A.id
from A
join B
on A.id = B.id
将会有一组mappers
读这两个表并基于连接键id
发送到reducers
,假设id=1
的行分发到Reducer R1
,id=2
的分发到R2
等等,这些reducer
对A
和B
进行交叉连接,R4
从A
得到id=4
的所有行,但是不会产生任何结果。
现在我们假定A
在id=1
上倾斜,这样R2
和R3
将会很快完成但是R1
会执行很长时间,因此成为job
的瓶颈。若是用户知道这些倾斜信息,这种瓶颈可以使用如下方法人工避免:
将查询分为两个部分:
select
A.id
from A
join B
on A.id = B.id
where A.id <> 1;
select
A.id
from A
join B
on A.id = B.id
where A.id = 1
and B.id = 1;
第一个查询没有倾斜数据将会很快的完成,如果我们假定表
B
中只有少量的B.id=1
的行,能够直接加载到内存中,通过将B
的数据存储到内存中的哈希表中,join
将会高效的完成,因此可以再mappper
端进行连接,而不用reduce
,效率会高很多,最后合并结果。
以上过程有如下缺点:
**** 1.表A
和表B
需要分别读和处理两次;
**** 2.结果需要合并;
**** 3.需要人工的处理倾斜数据。
hive为了避免上面的操作,在处理数据是对倾斜值进行特殊处理,首先读表B并且存储B.id=1
的数据到内存的哈希表中,运行一组mappers
来读取表A
并做以下操作:
- 若
id
为1
,使用B
的id=1
的哈希表来计算结果 - 对于其他值,发送到
reducer
端来join
,这个reduce
也会从B
的mapper
中得到对应需要连接的数据。
使用这种方法,最终我们只读取B
两次(存储倾斜值数据到哈希表和其他值map/reudce
,若是合并为读一遍的话,A
的map
要等表B
全处理完才能运行,就完全串行了,首先只处理倾斜数据到哈希表中,然后再B
,A map
并行,效率较高。我是这样理解的),并且A
中的倾斜数据在mapper
中进行连接,不会被发送到reducer
,其他的键值通过map/reduce
。
要注意到以上假设B
中的含有A
中倾斜值的行比较少,能够载入到内存中。
若是使用ListBucketing
对倾斜值单独存储,会有更好的性能。在读倾斜的数据到内存中时可以指定到倾斜目录下的数据。
ListBucketing
目标
有如下问题:存在许多表是这种格式
create table T
( a,
b,
c,
.....,
x
)
partitioned by (ds);
但是一下查询需要更加高效:
select ...
from T
where x = 10;
字段X
中含有倾斜数据,一般情况下x
的值中大约有10
个值有重度倾斜,其他值基数很小,当然,每天倾斜的X
的值可能改变,上边的文件可按照以下方法解决。
Basic Partitioning
可以将x
的所有值作为分区
create table T
(
a,
b,
c,
.......
)
partitioned by (ds, x)
但是这种办法会产生大量的目录,产生大量的文件,使用时需要注意小文件问题。
List Bucketing
上边方法提到将含有倾斜值得列作为分区存储,但是可能产生大量的目录,为什么不把列值不倾斜的放在一起呢,将每个倾斜的值单独存放一个目录,于是有了List Bucketing
。这个映射在表或分区级别的Metastore
中维护。倾斜键的列表存储在表级别中,这个列表可以由客户端周期的提供,并且在新的分区载入时可以被更新。
如下例子,一个表含有一个x
字段倾斜值的列表:6,20,30,40
。当一个新的分区载入时,它会创建5
个目录(4个目录对应x的4个倾斜值,另外一个目录是其余值
)。这个表的数据被分成了5
个部分:6,20,30,40,others
。这跟上一节介绍的分桶表类似,桶的个数决定了文件的个数
。倾斜键的整个列表存储在每个表或者分区中。
当使用一下查询时,hive
编译器会仅仅使用x=30
对应的目录去运行map-reduce
。
select ...
from T
where ds = '2012-04-15'
and x = 30;
若是查询是x=50
,则会使用x=others
对应的目录去运行map-reduce
。
select ...
from T
where ds = '2012-04-15'
and x = 50;
这种方法在一下条件下是很有效的:
- 每个分区的倾斜键占总数据的一大部分,在上边的例子中,如果倾斜的键(
6,20,30,40
)只占一小部分数据(比如20%
),那么在查询x=50
时依然需要扫描80%
的数据。 - 每个分区倾斜值得数据不能够太大,因为这个倾斜值列表存在在元数据库中,在元数据库中为每个分区存储
100w
个倾斜键是没有意义的。
这种方法也可被扩展到含有多个列产生的倾斜键,例如我们想优化一下查询
select ...
from T
where x = 10
and y = 'b';
扩展以上的方法,对于(x,y)
每个倾斜值,也按照上边方式单独存储,因此元数据库会有以下映射:(10, ‘a’) -> 1, (10, ‘b’) -> 2, (20, ‘c’) -> 3
,(others) -> 4
。
因此可直接找到2
对应的目录,减少要处理的数据。
同时一下查询也会有一定程度的优化:
select ...
from T
where x = 10;
select ...
from T
where y = 'b';
以上两个语句在执行的过程中会裁剪掉一部分数据,例如,对x=10
的查询hive编译器可以裁剪掉(20,c)
对应的文件,对于y=’b’,(10, ‘a’)
和(20, ‘c’)
对应的文件会被裁剪掉,一定程度能够减少扫描的数据量。这种方法不适用于一下场景:
- 倾斜键数目非常多,元数据规模问题
- 许多情况下,倾斜键由多个列组成,但是在查询中,没有使用到倾斜键中的那些列。
Skewed Table vs. List Bucketing Table
Skewed Table
是一个表它含有倾斜的信息。List Bucketing Table
是Skewed Table
,此外,它告诉hive
使用列表桶的特点:为倾斜值创建子目录。
以下说明两者的存储区别:
create table t1
(x string)
skewed by (x) on (‘a’, ‘b’)
partitioned by dt
location '/user/hive/warehouse/t1';
create table t2
(x string)
skewed by (x) on (‘a’, ‘b’)
STORED AS DIRECTORIES
partitioned by dt
location '/user/hive/warehouse/t2';
两者存储的形式如下所示:
/user/hive/warehouse/t1/dt=something/data.txt
/user/hive/warehouse/t2/dt=something/x=a/data.txt
/user/hive/warehouse/t2/dt=something/x=b/data.txt
/user/hive/warehouse/t2/dt=something/default/data.txt
List Bucketing Validation
由于列表桶的子目录特点,它不能够与一些特征共存。
(1)DDL
列表桶与以下共存会抛出编译错误:
normal bucketing (clustered by, tablesample, etc.)
external table
“load data …”
CTAS (Create Table As Select) queries
(2)DML
与一下DML
操作共存也会跳出错误:
“insert into”
normal bucketing (clustered by, tablesample, etc.)
external table
non-RCfile due to merge
non-partitioned table
以上是关于Hive SQL使用Skewed方式建表解决数据倾斜小记的主要内容,如果未能解决你的问题,请参考以下文章