HIVE Skewed Table

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 <T> (SCHEMA) SKEWED BY  (keys) ON ('c1', 'c2') [STORED AS DIRECTORIES];

这个语句仅支持表级别，不支持分区级别。它可以将一个普通表转为倾斜表，也可以修改倾斜表的倾斜列名称和值。它会影响到之后创建的分区，对alter语句之前的分区无影响。

ALTER TABLE <T> (SCHEMA) NOT SKEWED;

关闭倾斜特征，使一个表成为非倾斜表，同样会影响到之后创建的分区，对alter语句之前的分区无影响。

ALTER TABLE <T> (SCHEMA) NOT STORED AS DIRECTORIES;

关闭list bucketing，表成为普通的倾斜表。

ALTER TABLE <T> (SCHEMA) SET SKEWED LOCATION (key1="loc1", key2="loc2");

修改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;
（官网这样写，难道不是应该先where后join么？需要了解下）

第一个查询没有倾斜数据将会很快的完成，如果我们假定表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

参考文献

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