TUNANのBlogHomeArchivesTagsCategories
HIVE Skewed Table&List Bucketing
|Hive
Word count: 2.4k|Reading time: 8 min

目录

  1. HIVE Skewed Table
  2. Skewed Join Optimization(最优化)
  3. Basic Partitioning
  4. List Bucketing
  5. Skewed Table vs List Bucketing Table
  6. List Bucketing Validation

HIVE Skewed Table

Skewed Table可用于提高一个或多个列具有偏斜值的表的性能。通过指定经常出现的值（严重偏斜），Hive会自动将它们拆分成单独的文件（或在列表存储的情况下为目录），并在查询过程中考虑到这一事实，以便它可以跳过或包括整个文件（或在列表存储的情况下为目录）。可以在表创建过程中在每个表级别上指定。

若是指定了STORED AS DIRECTORIES，也就是使用列表桶（ListBucketing），hive会对倾斜的值建立子目录，查询会更加得到优化。

下面的示例显示具有三个偏斜值的一列，还可以选择使用STORED AS DIRECTORIES子句来指定列表存储。

CREATE TABLE list_bucket_single (key STRING, value STRING)
SKEWED BY (key) ON (1,5,6) [STORED AS DIRECTORIES];

这是一个带有两个倾斜列的表的示例。

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 table_name SKEWED BY (col_name1, col_name2, ...) ON ([(col_name1_value, col_name2_value, ...) [, (col_name1_value, col_name2_value), ...][STORED AS DIRECTORIES];

STORED AS DIRECTORIES选项确定倾斜表是否使用列表存储功能，该功能为倾斜值创建子目录。

ALTER TABLE table_name NOT SKEWED;

NOT SKEWED选项使表不倾斜，并关闭列表存储功能（因为列表存储表始终是倾斜的）。这会影响在ALTER语句之后创建的分区，但对在ALTER语句之前创建的分区没有影响。

ALTER TABLE table_name NOT STORED AS DIRECTORIES;

NOT STORED会关闭列表存储功能，但是表仍然歪斜。

ALTER TABLE table_name SET SKEWED LOCATION (col_name1="location1" [, col_name2="location2", ...] );

修改list bucketing倾斜值的存储位置映射。

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，效率会高很多，最后合并结果。

优点：

  • 如果少量的倾斜键占了很大一部分数据，它们将不会成为瓶颈。

缺点：

  • 表A和表B需要分别读和处理两次；

  • 结果需要合并；

  • 需要人工的处理倾斜数据。

hive为了避免上面的操作，在处理数据是对倾斜值进行特殊处理，首先读表B并且存储B.id=1的数据到内存的哈希表中，运行一组mappers来读取表A并做以下操作：

  1. 若id为1，使用B的id=1的哈希表来计算结果
  2. 对于其他值，发送到reducer端来join，这个reduce也会从B的mapper中得到对应需要连接的数据。

使用这种方法，最终我们只读取B两次，并且A中的倾斜数据在mapper中进行连接，不会被发送到reducer，其他的键值通过map/reduce。

假设B的行很少，而键在A中倾斜。因此可以将这些行加载到内存中。若是使用ListBucketing对倾斜值单独存储，会有更好的性能。在读倾斜的数据到内存中时可以指定到倾斜目录下的数据。

Basic Partitioning

有如下问题：存在许多表是这种格式

create table T(a, b, c, ....., x) partitioned by (ds);

但是以下查询需要更加高效(这不扯蛋吗，有分区的查询条件不是分区)：

select ... from T where x = 10;

字段x中含有倾斜数据，一般情况下x的值中大约有1000个值有重度倾斜，其他值基数很小，当然，每天倾斜的X的值可能改变，上述要求可以通过以下方式解决：

为值“ x”创建一个分区。

create table T(a,b,c, .......) partitioned by (ds, x)

优点

  • 现有的Hive就足够了。

缺点

  • HDFS可伸缩性：HDFS中的文件数量增加。
  • HDFS可伸缩性：HDFS中的中间文件数量增加。例如，如果有1000个映射器和1000个分区，并且每个映射器每个键至少获得1行，我们最终将创建100万个中间文件。
  • Metastore的可伸缩性：Metastore会随着分区数量的增长而扩展。

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;

这种方法在一下条件下是很有效的：

  1. 每个分区的倾斜键占总数据的一大部分。在上边的例子中，如果倾斜的键（6，20，30，40）只占一小部分数据（比如20%）,那么在查询x=50时依然需要扫描80%的数据。
  2. 每个分区的倾斜键数量非常少，因为这个倾斜值列表存在在元数据库中，在元数据库中为每个分区存储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

由于列表桶的子目录特点，它不能够与一些特征共存。

DDL

列表桶与以下共存会抛出编译错误：

  • normal bucketing (clustered by, tablesample, etc.)
  • external table
  • “ load data …”
  • CTAS (Create Table As Select) queries

DML

与一下DML操作共存也会跳出错误：

  • “ insert into ”
  • normal bucketing (clustered by, tablesample, etc.)
  • external table
  • non-RCfile due to merge
  • non-partitioned table

参考文献：

1.HIVE ListBucketing
2.HIVE LanguageManualDDL-SkewedTables
3.HIVE Skewed Join Optimization

Author: Tunan
Link: http://yerias.github.io/2018/11/10/hive/11/
Copyright Notice: All articles in this blog are licensed under CC BY-NC-SA 4.0 unless stating additionally.
Hive