Hive构建在Hadoop的HDFS和MapReduce之上，用于管理和查询结构化/非结构化数据的数据仓库。
使用HQL作为查询接口，使用HDFS作为底层存储，使用MapReduce作为执行层
用户接口：包括 CLI,JDBC,ODBC和 WUI
Hive内部执行流程：解释器、编译器、优化器、执行器
元数据存储。通常是存储在关系数据库如 mysql, derby 中
Hadoop：用 HDFS 进行存储，利用 MapReduce 进行计算
from … where … select … group by … having … order by … limit …
注意事项
mysql执行顺序
Text File format : 默认格式，数据不做压缩，磁盘开销大，数据解析开销大。
Sequence File format
面向行：在一起存储的同一行数据是连续存储
RCfile format : RCFILE 是一种行列存储相结合的存储方式。首先，其将数据按行分块，保证同一个 record 在一个块上，避免读一个记录需要读取多个 block。其次，块数据列式存储，有利于数据压缩和快速的列存取。RCFile 目前没有性能优势，只有存储上能省 10% 的空间。
Parquet :
AVRO : avro Schema 数据序列化。
ORC : 对RCFile做了一些优化，支持各种复杂的数据类型 性能比较好
小文件如何产生的
动态分区插入数据，产生大量的小文件，从而导致map数量剧增；
倒入数据时产生，每执行一次 insert 时hive中至少产生一个文件，文件数量=MapTask数量*分区数，insert 导入时至少会有一个MapTask。像有的业务需要每10分钟就要把数据同步到 hive 中，这样产生的文件就会很多。
-- 通过load方式加载数据
load data local inpath '/export/score' overwrite into table A -- 导入文件夹
-- 通过查询方式加载数据
insert overwrite table A select s_id,c_name,s_score from B;
reduce数量越多，小文件也越多（reduce的个数和输出文件是对应的）；
数据源本身就包含大量的小文件。
造成的影响
如何解决
使用 hive 自带的 concatenate 命令，自动合并小文件 alter table A concatenate;
一个节点上split的至少的大小(这个值决定了多个DataNode上的文件是否需要合并)
set mapred.min.split.size.per.node
-- 设置map输入合并小文件的相关参数：
set mapred.min.split.size.per.node
-- 每个Map最小输入大小(这个值决定了合并后文件的数量)
set mapred.min.split.size=256000000;
– 设置map端输出进行合并，默认为true
set hive.merge.mapfiles = true;
– 设置reduce端输出进行合并，默认为false
set hive.merge.mapredfiles = true
– hive的查询结果输出是否进行压缩
set hive.exec.compress.output=true;
– MapReduce Job的结果输出是否使用压缩
set mapreduce.output.fileoutputformat.compress=true;
使用Sequencefile作为表存储格式，不要用textfile，在一定程度上可以减少小文件；
减少reduce的数量（可以使用参数进行控制）；set mapreduce.job.reduces=10;
少用动态分区，用时记得按distribute by分区；
-- map端聚合
set hive.map.aggr=true
-- map端自动负载均衡
set hive.groupby.skewindata = true
-- 小文件合并
set mapred.min.split.size= 256000000 -- 256M
-- 设置reduce个数
set mapred.reduce.tasks=10
-- 开启严格模式
set hive.mapred.mode = strict
hive.optimize.cp
hive.optimize.pruner
SELECT * FROM stu as t
LEFT JOIN course as t1
ON t.id=t2.stu_id
WHERE t.age=18;
上面语句是否具有优化的空间？如何优化？
SELECT FROM (SELECT FROM stu WHERE age=18) as t
LEFT JOIN course AS t1 on t.id=t1.stu_id
解决方案：
map段部分聚合
有数据倾斜的时候进行负载均衡（默认是false）
set hive.groupby.skewindata = true
阶段拆分-两阶段聚合 需要聚合的key前加一个随机数的前后缀，这样就均匀了，之后再按照原始的key聚合一次
生成的查询计划有两 个 MapReduce 任务。在第一个 MapReduce 中，map 的输出结果集合会随机分布到 reduce 中， 每个 reduce 做部分聚合操作，并输出结果。相同的 GroupB Key 有可能分发到不同的 reduce 中，从而达到负载均衡的目的；第二个 MapReduce 任务再根据预处 理的数据结果按照 Group By Key 分布到 reduce 中（这个过程可以保证相同的 Group By Key 分布到同一个 reduce 中），最后完成最终的聚合操作。
假设 key = 水果
select count(substr(a.tmp,1,2)) as key
from(
select concat(key,'_',cast(round(10*rand())+1 as string)) tmp
from table
group by tmp
)a
group by key
map的数量不是越多越好，如果一个任务有很多小文件（远远小于块大小128m），则每个小文件也会被当做一个块，用一个map任务来完成，而一个map任务启动和初始化的时间远远大于逻辑处理的时间，就会造成很大的资源浪费 。而且，同时可执行的map数是受限的
mapred.min.split.size: 指的是数据的最小分割单元大小；min的默认值是1B
mapred.max.split.size: max的默认值是256MB
小文件问题：
当input的文件任务逻辑复杂，map执行非常慢的时候，可以考虑增加Map数，来使得每个map处理的数据量减少，从而提高任务的执行效率。
启动和初始化reduce也会消耗时间和资源；
另外，有多少个reduce，就会有个多少个输出文件，如果生成了很多个小文件，那么如果这些小文件作为下一个任务的输入，则也会出现小文件过多的问题；
如果Reduce设置的过小，那么单个Reduce处理的数据将会加大，很可能会引起OOM异常
处理大数据量利用合适的Reduce数；使单个Reduce任务处理数据量大小要合适；
set mapred.reduce.tasks=10; 就是10个 如果是-1 就会估算
调整hive.exec.reducers.bytes.per.reducer参数的值 每个reduce处理数据量；
什么情况下只有一个reduce；
select uid,upload_time,event_type,record_data
from calendar_record_log
where pt_date >= 20190201 and pt_date <= 20190224
distribute by uid
sort by upload_time desc,event_type desc;
原因：distinct会将列中所有的数据保存到内存中 ，极有可能发生内存溢出
采用sum() group by的方式来替换count(distinct) 完成计算。
解决方案 ：可以考虑使用Group By 或者 ROW_NUMBER() OVER(PARTITION BY col) 方式代替COUNT(DISTINCT col)
select count(distinct a) from calendar_record_log
where pt_date >= 20190101;
– 但是这样写会启动两个MR job（单纯distinct只会启动一个），
– 所以要确保数据量大到启动job的overhead远小于计算耗时，才考虑这种方法。
select count(1) from (
select uid from calendar_record_log
where pt_date >= 20190101
group by uid
) t;
– 用group by方式同时统计多个列？下面是解决方法：
select t.a,sum(t.b),count(t.c),count(t.d) from (
select a,b,null c,null d from some_table
union all
select a,0 b,c,null d from some_table group by a,c
union all
select a,0 b,null c,d from some_table group by a,d
) t;
处理掉字段中带有空值的数据
原因：一个表内有许多空值时会导致MapReduce过程中,空成为一个key值,对应的会有大量的value值, 而一个key的value会一起到达reduce造成内存不足
1.在查询的时候，过滤掉所有为NULL的数据，比如：
create table res_tbl as
select n.* from
(select * from res where id is not null ) n
left join org_tbl o on n.id = o.id;
2.查询出空值并给其赋上随机数,避免了key值为空（数据倾斜中常用的一种技巧）
create table res_tbl as
select n.* from res n
full join org_tbl o
on case when n.id is null then concat(‘hive’, rand()) else n.id end = o.id;
单独处理倾斜key
不同数据类型，这种情况不太常见，主要出现在相同业务含义的列发生过逻辑上的变化时。不转换类型，计算key的hash值时默认是以int型做的，这就导致所有“真正的”string型key都分配到一个reducer上。所以要注意类型转换
select a.uid,a.event_type,b.record_data
from calendar_record_log a
left outer join (
select uid,event_type from calendar_record_log_2
where pt_date = 20190228
) b on a.uid = b.uid and b.event_type = cast(a.event_type as string)
where a.pt_date = 20190228
set hive.exec.mode.local.auto=true; //开启本地mr
set hive.exec.parallel=true //可以开启并发执行。
Hive 中数据倾斜的基本表现
如何产生
如何解决
原因：
解决方案：
-- 创建静态分区 数据加载到指定的分区
create table if not exists part1(
uid int,
uname string,
uage int
)PARTITION BY (country string)
row format delimiterd fileds terminated by ',';
(stored as ORC| SequenceFile) ORC、 SequenceFile都是存储方式
(loacation 地址)
– 导入数据 需要指定分区 数据未知，根据分区值确定创建分区
load data local inpath ‘/usr/loacl/xxx’
into table part1 partition(country=‘China’);
– 开启动态分区 默认为false，不开启
set hive.exec.dynamic.partition=true;
hive.exec.dynamic.parition.mode=nonstrict;
– 创建动态双分区
create table if not exists dt_part1(
uid int,
uname string,
uage int
)
PARTITIONED BY (year string,month string)
row format delimited fields terminated by ‘,’;
– 在文件系统中的表现为date_time为一个文件夹，type为date_time的子文件夹。
– 追加写入数据
insert into dy_part1 partition(year,month)
select from part_tmp;
– 覆盖写入数据
insert overwrite dy_part1 partition(year,month)
select from part_tmp;
– 混合分区
create table if not exists dy_part2(
uid int,
uname string,
uage int
)
PARTITIONED BY (year string,month string)
row format delimited fields terminated by ‘,’;
– 插入数据
insert into dy_part2 partition(year=‘2018’,month)
select uid,uname,uage,month from part_tmp；
– 多个范围分区键
create table test_demo (value int)
partitioned by range (id1 INT, id2 INT, id3 INT)
(
– id1在(–∞,5]之间，id2在(-∞,105]之间，id3在(-∞,205]之间
partition p5_105_205 VALUES LESS THAN (5, 105, 205),
– id1在(–∞,5]之间，id2在(-∞,115]之间，id3在(-∞,+∞]之间
partition p5_115_max VALUES LESS THAN (5, 115, MAXVALUE)
)
– 查看分区数据
select * from part1 where country = ‘China’;
– 显示分区
show partitions part1;
– 增加分区
alter table part1 add partition(country = ‘india’) partition(country = ‘America’);
– 增加分区并设置数据
alter table part1 add partition(country = ‘xxx’)
location ‘user/hive/warehouse/xxx’
– 修改分区的存储路径 hdfs路径必须是全路径
alter table part1 partition(country=‘Vietnam’)
set location ‘hdfs://hadoop01:9000/user/hive/warehouse/brz.db/part1/country=Vietnam’
– 删除分区
alter table part1 drop partition(country = ‘india’)
– 手动向hdfs中创建分区目录，添加数据，创建好hive的外表之后，无法加载数据，
– 元数据中没有相应的记录
msck repair table tablename
定义：
导入数据
导入数据有两种，一种是通过文件导入，但是并不会真正的分桶,load data只是把文件上传到 表所在的HDFS目录下。并没有做其他操作 ；一种是通过从其他表插入的方式导入数据，这种方式才能真正的分桶 insert … select；
cluster by (uid) – 指定getPartition以哪个字段来进行hash散列，并且排序字段也是指定的字段，默认以正序进行排序
distribute by(uid) – 指定getPartition以哪个字段来进行hash散列
sort by(uid asc) – 指定排序字段，以及排序规则，更灵活的方式，这种数据获取方式可以分别指定getPartition的字段和sort的字段
方式1
方式2
cluster by (uid)与distribute by(uid) sort by (uid asc)结果是一样的
抽样语句 ：tablesample（bucket x out of y）
y必须是table总共bucket数的倍数或者因子。
例如：table总共分了64份，当y=32时，抽取2(64/32)个bucket的数据，当y=128时，抽取1/2(64/128)个bucket的数据。x表示从哪个bucket开始抽取。
例如：table总共bucket数为32，tablesample(bucket 3 out of 16)表示总共抽取2(32/16)个bucket的数据，分别为第3个bucket和第19(3+16)个bucket的数据。
select * from table_name tablesample(n percent) 抽出n%的数据 全表扫描
如果在 TABLESAMPLE 子句中指定的列与 CLUSTERED BY 子句中的列相匹配，则 TABLESAMPLE 只扫描表中要求的哈希分区【就是具体的桶】
--创建一个分桶表 并且指定排序字段及排序规则
create table if not exists buc1(
uid int,
uname string,
uage int
)
distribute by (uid)
sorted by(uid desc) into 4 buckets
row format delimited fields terminated by ',';
– cluster by (uid)指定getPartition以哪个字段来进行hash散列，并且排序字段也是指定的字段，默认以正序进行排序
– distribute by(uid) – 指定getPartition以哪个字段来进行hash散列
– 加载数据 方式1
– 打开enforce bucketing开关,设置强制分桶属性
set hive.enforce.bucketing=true
set mapred.reduce.tasks = -1
insert overwrite table buc1
select uid,uname,uage from buc_temp
sort by (uid);
– 加载数据 方式2
– 将reducer个数设置为目标表的桶数，并在 SELECT 语句中用 DISTRIBUTE BY <bucket_key>
– 对查询结果按目标表的分桶键分进reducer中。
set hive.enforce.bucketing = false
set mapred.reduce.tasks = num_buckets
insert into table buc1
select uid,uname,uage from buc_temp
distribute by (uid) sort by (uage desc);
– 查看表结构
desc formatted tablename;
– 分桶查询结果
select from buc1 cluster by (uid);
‘’’
采样 TABLESAMPLE(BUCKET x OUT OF y)
x：表示从哪个 bucket 开始抽取数据 y：必须为该表总 bucket 数的倍数或因子
‘’’
– 查询第几桶 取出 uid % 4 == 0的数据
select from buc1 tablesample(bucket 1 out of 4 on uid);
– 查询uid 为奇数
select from buc1 tablesample(bucket 2 out of 2 on uid)
– 随机查询三条数据
select from part_tmp order by rand() limit 3;
select * from part_tmp tablesample(0.1 percent) ;
-- 按照性别进行分区（1男2女），在分区中按照uid的奇偶进行分桶：
-- 分区使用的是表外字段，分桶使用的是表内字段
1 gyy1 1
2 gyy2 2
3 gyy3 2
4 gyy4 1
5 gyy5 2
6 gyy6 1
7 gyy7 1
8 gyy8 2
9 gyy9 1
10 gyy10 1
11 gyy11 2
12 gyy12 1
– 创建带有分区的分桶表
create table if not exists stus(
uid int,
uname string
)
partitioned by(sex int)
clustered by(uid) into 2 buckets
row format delimited filed terminated by ’ ‘;
– 创建临时表
create table if not exists stu_temp(
uid int,
uname string,
usex int
)
row format delimited fields terminated by ’ ';
– 临时表中添加数据
load data local inpath ‘/usr/local/hivedata/stu.dat’ into table stu_temp
– 分桶表中加数据
insert into table stus partition(sex)
select uid,uname,usex from stu_temp
cluster by (uid);
– 查询性别为女性的、并且学号为奇数的学生：
select * from stus tablesample(bucket 2 out of 2 on uid)
where sex=2;
关系函数： <= 、 >= 、 IS NULL 、IS NOT NULL、LIKE
日期函数： to_date、 year 、month 、second 、weekofyear、 datediff时间比较
条件函数： IF CASE
字符串函数：length、 reverse、 substr 截取字符串、lower、 trim去空格、CONCAT 字符串拼接
统计函数：
Hive的SQL还可以通过用户定义的函数（UDF），用户定义的聚合（UDAF）和用户定义的表函数（UDTF）进行扩展。
UDF、UDAF、UDTF的区别：
split将字符串转化为数组，即：split(‘a,b,c,d’ , ‘,’) ==> [“a”,“b”,“c”,“d”]。
coalesce(T v1, T v2, …) 返回参数中的第一个非空值；如果所有值都为 NULL，那么返回NULL。
collect_list列出该字段所有的值，不去重 select collect_list(id) from table。