标签：倾斜分桶 -- 分区高级操作数据 id hive03

Hive 分区表

https://blog.csdn.net/weixin_41122339/article/details/81584110
表在存储时，可以将数据根据分区键的列值存储在表目录的子目录中。这样将数据切分到不同目录存储后，可以加快对分区键字段的查询和过滤速度，通过在查询条件中指定过滤条件，可以只对指定目录的数据进行扫描，避免了全表扫描。
分区是以字段的形式在表结构中存在，通过 describe table 命令可以查看到字段存在，但是该字段不存放实际的数据内容，仅仅是分区的表示。
分区的目的是将一些数据分散到特定的子目录中，查询时就只选择查询某些子目录中的数据，以此加快查询效率。

入门操作

静态分区

-- 创建分区表并指定分区键
create table tb1(
    name string,
    age int
)
partitioned by (dt string)
stored as textfile;

-- 单条插入数据，但是不推荐，会生成大量小文件
insert into tb1 
partition (dt='2020-11-08')
values('zs', 18);

分区表创建成功后，就可以向分区表中插入数据了，根据插入数据时存放的分区，可以将分区分为静态分区和动态分区。

静态分区在导入数据时必须通过 partition (xxx=xxx) 显式指定分区字段名。

-- 静态分区在数据导入时必须手动指定目标分区
insert into tbl
partition (dt='2020-11-08')
values('1s', 25);

-- 创建分区表2，并将分区表1的数据导入分区表2
create table tb2(
name string,
age int
)
partitioned by (dt string)
stored as textfile;

insert into table tb2
partition (dt='2020-11-08')
select name,age from tb1;

// 导入外部文件来加载数据
load data local inpath '/home/hadoop/datas/dept/detp_01.txt' into table tb2 partition(dt='xxxxx');

外部表同样可以创建分区表，只需要将文件上传到 hdfs 指定的目录下就可以：

create external table employees_ex
 (
 name   string,
 salary  float,
 subordinated array<string>,
 deductions map<string,float>,
 address  struct<street:string,city:string,state:string,zip:int>
 )
 partitioned by (country string,state string)
 row format delimited
 fields terminated by "\t"
 collection items terminated by ","
 map keys terminated by ":"
 lines terminated by "\n"
 stored as textfile;   
 location "/user/had/data/"    //他其实和普通的静态分区表一样就是多了一个external关键字

动态分区

如果想要在数据导入时，由系统帮助判断数据属于哪个分区，可以使用动态分区，但是需要配置开启。

# 开启动态分区功能
set hive.exec.dynamic.partition=true;

# 指定动态分区模式
set hive.exec.dynamic.partition.mode=nostrict;

# 其他配置参数
set hive.exec.max.dynamic.partitions.pernode; --每一个执行mr节点上，允许创建的动态分区的最大数量(100)
set hive.exec.max.dynamic.partitions;  --所有执行mr节点上，允许创建的所有动态分区的最大数量(1000)
set hive.exec.max.created.files;  --所有的mr job允许创建的文件的最大数量(100000)
hive.error.on.empty.partition; --当有空分区生成时,是否抛出异常(false)

-- 开启动态分区配置
set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nostrict;

-- 对动态分区插入数据
insert into table tbl
partition (dt)
select * from tb2;

分桶

分区是为了将数据存储到不同的子目录中，而分桶是一种改变表的存储模式，从而完成对表优化的一种调优方式。
分桶是将表拆分到了不同文件中进行存储。通过改变数据的存储分布，提升查询、取样、Join 等任务执行效率。最常见的分桶操作就是对 Key 哈希取模，这样可以直接确定数据存放的桶文件。
另外在 Hive 中两张表如果需要执行 join 操作，转换为 MapReduce/Spark 任务后，join 操作一定会涉及到 Shuffle，而 Shuffle 过程耗费大量时间，如果以关键字作为分桶键，相同 key 一定会定位到相同的桶文件，那么可以直接对两张表的桶文件进行 join 处理，以此提升效率。

分桶表操作

SQL中使用 DISTRIBUTE BY / CLUSTER BY 指定分桶键完成数据分桶写入

CREATE [EXTERNAL] TABLE <table_name>
    (<col_name> <data_type> [, <col_name> <data_type> ...])]
    [PARTITIONED BY ...]            // 分区
    CLUSTERED BY (<col_name>)
        [SORTED BY (<col_name> [ASC|DESC] [, <col_name> [ASC|DESC]...])] 
        INTO <num_buckets> BUCKETS  // 分桶  
    [ROW FORMAT <row_format>] 
    [LOCATION '<file_path>']    
    [TBLPROPERTIES ('<property_name>'='<property_value>', ...)];

-- 指定分桶键并倒序排序
INSERT INTO TABLE tb_buckets_desc
    SELECT * FROM tb_buckets
        DISTRIBUTE BY id SORT BY id desc;

-- 设置分桶数并按照分桶键升序排列
set mapred.reduce.tasks=3;

INSERT INTO TABLE tb_buckets_desc
    SELECT * FROM tb_buckets
        DISTRIBUTE BY id SORT BY id asc;

INSERT INTO TABLE tb_buckets_desc
    SELECT * FROM tb_buckets
        CLUSTER BY id;

Hive 倾斜表、事务表

倾斜表

对于一列或者多列出现倾斜值的表，可以创建倾斜表来进一步提升性能。比如表中 Key 字段包括的数据，有大部分为同一数据值，这就造成了数据倾斜。数据倾斜会导致数据处理时，部分节点 Reduce 任务完成，但是需要等待倾斜部分处理完成。

倾斜表使用

// 创建单字段倾斜表
CREATE TABLE skewed_single (key string, value string)
SKEWED BY (key) ON ('1', '5', '6')
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

// 创建多字段的倾斜表
CREATE TABLE skewed_multiple (key STRING, value STRING)
SKEWED BY (key, value) ON (('1','One'), ('3','Three'), ('5','Five'))
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';

倾斜表创建后，会产生倾斜连接优化的效果。假设表 A id字段有值（1,2,3,4），表 B id字段有值（1,2,3），如果我们执行 join 操作：
select A.id from A join B on A.id = B.id;

此时会有一组 mappers 读取两个表并基于连接键 id 发送到 reducers，如果我们在 id=1 上倾斜，那么 Reducer2、Reducer3 会很快完成，Reducer1 会长时间执行，成为性能瓶颈。

如果能提前知道哪部分数据倾斜，可以分成两个部分执行 join 操作：

--首先进行Join，排除倾斜数据A.id=1
select A.id from A join B on A.id = B.id where A.id <> 1;
--单独对倾斜数据进行Join
select A.id from A join B on A.id = B.id where A.id = 1 and B.id = 1;

但是如果设置了倾斜表，会自动进行下面的优化：

将 B 表中 id=1 的数据加载到内存哈希表中，然后分发到 A 表的所有 Mapper 任务中，直接与 A.id=1 的数据进行 join；
其它非倾斜数据，则执行普通的 Reduce 操作，并进行 join；

List Bucketing

List Bucketing 是特殊的倾斜表，它会将倾斜的数据保存到单独的目录，这样与其他数据分开保存，提升查询、join 效率。

// 创建倾斜表，并指定 List Bucketing
CREATE TABLE list_bucket_single (key STRING, value STRING)
partitioned by (dt string)
  SKEWED BY (key) ON (1,5,6) STORED AS DIRECTORIES
STORED AS ORCFile;

// 导入数据
insert OVERWRITE table list_bucket_single
partition (dt='2020-11-19')
select * from skewed_single;

可以发现将倾斜字段（1，5，6）和其他字段分为不同目录存储。

倾斜表 DDL 操作

// 转换倾斜表为普通表
ALTER TABLE <T> (SCHEMA) SKEWED BY  (keys) ON ('c1', 'c2') [STORED AS DIRECTORIES];

// 修改普通表为倾斜表
ALTER TABLE <T> (SCHEMA) NOT SKEWED;

// 转换 ListBucketing 为普通倾斜表
ALTER TABLE <T> (SCHEMA) NOT STORED AS DIRECTORIES;

// 更改倾斜值的存放目录
ALTER TABLE <T> (SCHEMA) SET SKEWED LOCATION (key1="loc1", key2="loc2");
--将list_bucket_multiple表的组合倾斜值('1','One')、('3','Three')存放目录进行更改
ALTER TABLE list_bucket_multiple SET SKEWED LOCATION (('1','One')="hdfs://node01:9000/file/one_data" ,('3','Three')="hdfs://node01:9000/file/one_data");

事务表

hive 支持行级别的 ACID 语义，意味着可以同时向一个分区数据插入、查询、删除等操作。
事务功能仅支持 ORC 表存储格式，而且依赖分桶的存储格式，所以事务必须进行分桶操作。

// 创建事务表，分桶、开启事务、指定存储格式
CREATE TABLE employee (id int, name string, salary int)
CLUSTERED BY (id) INTO 2 BUCKETS
STORED AS ORC 
TBLPROPERTIES ('transactional' = 'true');

INSERT INTO employee VALUES
(1, 'Jerry', 5000),
(2, 'Tom',   8000),
(3, 'Kate',  6000);

查看初始数据存储状态：

HDFS 上的事务表包含两种类型文件（base文件、delta文件）：

base文件：用来存放平常数据，会定期执行任务将 delta 文件合并到 base 文件；
delta文件：用来存储新增、更新、删除的数据，每一个事务处理都会单独新建一个 delta 目录用来存储数据，目录下存储的数据按照桶划分。

两次事务操作后的表数据内容：

标签：倾斜,分桶,--,分区,高级,操作,数据,id,hive03
From： https://www.cnblogs.com/istitches/p/18326074

hive03_高级操作