hive(七) -- 拉链表、数据存储及优化配置

使用2号的数据 直接将1号的数据覆盖掉
优点:实现最简单，使用起来最方便
缺点:没有历史状态 想查询008之前的数据查看不到

1号创建一张表拉取所有数据
2号再创建一张表拉取所有数据 
... 每天都创建一张表
优点:记录了所有数据在不同时间的状态
缺点:冗余存储了很多没有发生变化的数据，导致存储的数据量过大

-- 数据准备
vi zipper.txt
001	186xxxx1234	laoda	0	sh	2021-01-01	9999-12-31
002	186xxxx1235	laoer	1	bj	2021-01-01	9999-12-31
003	186xxxx1236	laosan	0	sz	2021-01-01	9999-12-31
004	186xxxx1237	laosi	1	gz	2021-01-01	9999-12-31
005	186xxxx1238	laowu	0	sh	2021-01-01	9999-12-31
006	186xxxx1239	laoliu	1	bj	2021-01-01	9999-12-31
007	186xxxx1240	laoqi	0	sz	2021-01-01	9999-12-31
008	186xxxx1241	laoba	1	gz	2021-01-01	9999-12-31
009	186xxxx1242	laojiu	0	sh	2021-01-01	9999-12-31
010	186xxxx1243	laoshi	1	bj	2021-01-01	9999-12-31

--创建拉链表
create table dw_zipper
(
    userid    string,
    phone     string,
    nick      string,
    gender    int,
    addr      string,
    starttime string,
    endtime   string
) row format delimited fields terminated by '\t';
load data local inpath '/root/zipper.txt' into table dw_zipper;
select * from dw_zipper;

vi update.txt
008	186xxxx1241	laoba	1	sh	2021-01-02	9999-12-31
011	186xxxx1244	laoshi	1	jx	2021-01-02	9999-12-31
012	186xxxx1245	laoshi	0	zj	2021-01-02	9999-12-31

create table ods_update
(
    userid    string,
    phone     string,
    nick      string,
    gender    int,
    addr      string,
    starttime string,
    endtime   string
) row format delimited fields terminated by '\t';

load data local inpath '/root/update.txt' overwrite into table ods_update;

select * from ods_update;

create table tmp_zipper
(
    userid    string,
    phone     string,
    nick      string,
    gender    int,
    addr      string,
    starttime string,
    endtime   string
) row format delimited fields terminated by '\t';

insert overwrite table tmp_zipper
select
    userid,
    phone,
    nick,
    gender,
    addr,
    starttime,
    endtime
from ods_update
union all
--查询原来拉链表的所有数据，并将这次需要更新的数据的endTime更改为更新值的startTime
select
    a.userid,
    a.phone,
    a.nick,
    a.gender,
    a.addr,
    a.starttime,
    --如果这条数据没有更新或者这条数据不是要更改的数据，就保留原来的值，否则就改为新数据的开始时间-1
    if(b.userid is null or a.endtime < '9999-12-31', a.endtime , date_sub(b.starttime,1)) as endtime
from dw_zipper a  left join ods_update b
                            on a.userid = b.userid ;

insert overwrite table dw_zipper
select * from tmp_zipper;

假设我们查询sid列 
	行存储查询时会将课程名和年级一起查询出来,这样会造成数据冗余,如果列数比较多,影响比较大.
	列存储是按照列来存储到一起的,每一列单独存放,查询起来方便很多,直接查询即可

 行存储查询所有数据直接查询就可以了
 列存储,由于每列单独存放,先查询出来一列,然后在通过聚集运算,把列上每个数据拼接成行,没有行存储方便.

TextFile是Hive中默认的文件格式，存储形式为按行存储。工作中最常见的数据文件格式就是TextFile文件，几乎所有的原始数据生成都是TextFile格式.
建表时不指定存储格式即为textfile，导入数据时把数据文件拷贝至hdfs不进行处理。
优点 
    1.最简单的数据格式，不需要经过处理，可以直接cat查看
    2.可以使用任意的分隔符进行分割
    3.便于和其他工具（Pig, grep, sed, awk）共享数据
    4.可以搭配Gzip、Bzip2、Snappy等压缩一起使用
缺点 
	1.耗费存储空间，I/O性能较低
	2.结合压缩时Hive不进行数据切分合并，不能进行并行操作，查询效率低
	3.按行存储，读取列的性能差
应用场景
	1.适合于小量数据的存储查询
	2.一般用于做第一层数据加载和测试使用

Parquet是一种支持嵌套结构的列式存储文件格式.是一种支持嵌套数据模型对的列式存储系统，作为大数据系统中OLAP查询的优化方案，它已经被多种查询引擎原生支持，并且部分高性能引擎将其作为默认的文件存储格式。通过数据编码和压缩，以及映射下推和谓词下推功能，Parquet的性能也较之其它文件格式有所提升。
优点 
	1.更高效的压缩和编码可压缩、可分割，优化磁盘利用率和I/O
	2.可用于多种数据处理框架
缺点
	1.不支持update, insert, delete, ACID
应用场景
	1.适用于字段数非常多，无更新，只取部分列的查询

ORC（OptimizedRC File）文件格式也是一种Hadoop生态圈中的列式存储格式.它并不是一个单纯的列式存储格式,是首先根据行组分割整个表，在每一个行组内进行按列存储。.

Index Data
Index Data 包括每一列的最小值和最大值，以及每一列中的行位置。(也可以包含bit field or bloom filter ) 行索引项提供了偏移量，使您能够在解压缩块中查找正确的压缩块和字节。
注意，ORC索引仅用于选择stripes 和 row groups.
拥有相对频繁的行索引项可以在stripe 内跳过行，以便快速读取，尽管stripe 很大。默认情况下，可以跳过每10,000行。

Row Data
存的是具体的数据，先取部分行，然后对这些行按列进行存储。对每个列进行了编码，分成多个Stream来存储。

Stripe Footer
存的是各个Stream的类型，长度等信息。

File Footer
每个文件有一个File Footer，这里面存的是每个Stripe的行数，每个Column的数据类型信息等；

PostScript
每个文件的尾部是一个PostScript，这里面记录了整个文件的压缩类型以及FileFooter的长度信息等。

在读取文件时，会seek到文件尾部读PostScript，从里面解析到File Footer长度，再读FileFooter，从里面解析到各个Stripe信息，再读各个Stripe，即从后往前读。

优点
	1.列式存储，存储效率非常高
	2.可压缩，高效的列存取
	3.查询效率较高，支持索引
	4.支持各种复杂的数据类型
	5.支持矢量化查询
缺点
	1.加载时性能消耗较大
	2.需要通过text文件转化生成
	3.读取全量数据时性能较差
应用场景
	适用于Hive中大型的存储、查询

Hive的默认查询执行引擎一次处理一行，而矢量化查询执行是一种Hive针对ORC文件操作的特性，目的是按照每批1024行读取数据，并且一次性对整个记录整合（而不是对单条记录）应用操作，提升了像过滤, 联合, 聚合等等操作的性能。
注意：要使用矢量化查询执行，就必须以ORC格式存储数据。
-- 开启矢量化查询
set hive.vectorized.execution.enabled = true;
set hive.vectorized.execution.reduce.enabled = true;

create table test_textfile
(
    stime      string,
    userid     string,
    keyword    string,
    clickorder string,
    url        string
)row format delimited fields terminated by '\t';
load data local inpath '/root/sogou.txt' into table test_textfile;

select * from test_textfile limit 10;

desc formatted  test_textfile;

create table test_orc
(
    stime      string,
    userid     string,
    keyword    string,
    clickorder string,
    url        string
)
stored as ORC ;
-- 不能使用load加载数据 
insert into table test_orc select * from test_textfile ;

select * from test_orc limit 10;

create table test_parquet
(
    stime      string,
    userid     string,
    keyword    string,
    clickorder string,
    url        string
)
    stored as parquet  ;
-- 不能使用load加载数据 
insert into table test_parquet select * from test_textfile ;

select * from test_parquet limit 10;

desc formatted test_parquet;

压缩的优点
	减小文件存储所占空间
	加快文件传输效率，从而提高系统的处理速度
	降低IO读写的次数
压缩的缺点
	使用数据时需要先对文件解压，加重CPU负荷，压缩算法越复杂，解压时间越长

临时配置

--配置MapReduce开启输出压缩及配置压缩类型
--开启输出压缩
set mapreduce.output.fileoutputformat.compress=true;
--配置压缩类型为Snappy
set mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.SnappyCodec;

--配置Hive开启中间结果压缩和输出压缩及配置压缩类型
-- 中间结果压缩
set hive.exec.compress.intermediate=true;
set hive.intermediate.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;
-- 输出结果压缩
set hive.exec.compress.output=true;

永久配置
将以上MapReduce的配置写入mapred-site.xml中，重启Hadoop
将以上Hive的配置写入hive-site.xml中，重启Hive

create table t_textfile_snappy
    stored as textfile
as select * from test_textfile;

create table t_orc_snappy
    stored as orc tblproperties ("orc.compress"="SNAPPY")
as select * from test_textfile;

select * from t_orc_snappy limit 10;

explain  select * from test_textfile limit 10;

explain  select stime,count(*) from test_textfile where stime > '00:00:00' group by  stime order by  stime limit  5;

explain  select * from t_category left join t_product tp on t_category.cid = tp.cid;

尽量使用Map Join来实现Join过程，Hive中默认自动开启了Map Join：
hive.auto.convert.join=true
-- 如果大小表连接时小表数据小于该值,则自动开启mapjoin优化
-- 2.0以前
hive.smalltable.filesize or hive.mapjoin.smalltable.filesize: 默认值是25m
-- 2.0以后
hive.auto.convert.join.noconditionaltask.size :默认值10m
和
hive.auto.convert.join.noconditionaltask: 默认值是true

使用Bucket Join
语法：clustered by colName
参数：set hive.optimize.bucketmapjoin = true;
要求：分桶字段 = Join字段 ，桶的个数相等或者成倍数

使用Sort Merge Bucket Join（SMB）
基于有序的数据Join
语法：clustered by colName sorted by (colName)
参数
set hive.optimize.bucketmapjoin = true;
set hive.auto.convert.sortmerge.join=true;
set hive.optimize.bucketmapjoin.sortedmerge = true;
set hive.auto.convert.sortmerge.join.noconditionaltask=true;
要求：分桶字段 = Join字段 = 排序字段 ，桶的个数相等或者成倍数

hive.map.aggr=true;
通过减少shuffle数据量和Reducer阶段的执行时间，避免每个Task数据差异过大导致数据倾斜

需要注意:求平均值不能使用  combiner

hive.groupby.skewindata=true;
开启该参数以后，当前程序会自动通过两个MapReduce来运行
第一个MapReduce自动进行随机分布到Reducer中，每个Reducer做部分聚合操作，输出结果
第二个MapReduce将上一步聚合的结果再按照业务（group by key）进行处理，保证相同的分布到一起，最终聚合得到结果

键随机数打散

select a.id,a.value1,b.value2 from table1
		a join (select b.* from table2 b where b.ds>='20181201' and b.ds<'20190101') c
        on (a.id=c.id)

方案二：使用Bucket Join
如果使用方案一，过滤后的数据依旧是一张大表，那么最后的Join依旧是一个Reduce Join
这种场景下，可以将两张表的数据构建为桶表，实现Bucket Map Join，避免数据倾斜.

Skew Join是Hive中一种专门为了避免数据倾斜而设计的特殊的Join过程
这种Join的原理是将Map Join和Reduce Join进行合并，如果某个值出现了数据倾斜，就会将产生数据倾斜的数据单独使用Map Join来实现
其他没有产生数据倾斜的数据由Reduce Join来实现，这样就避免了Reduce Join中产生数据倾斜的问题
最终将Map Join的结果和Reduce Join的结果进行Union合并

hive.optimize.skewjoin = true，