数据分析系列（一）：Hive_hive数据分析

Hive

1. 简介

Hive是基于Hadoop的一个数据仓库工具，可以将结构化的数据文件映射为一张数据库表，并提供完整的SQL查询功能，可以将SQL语句转换为MapReduce任务运行。
主要的使用场景：非实时的、离线的、对响应及时性要求不高的海量数据批量计算，即席（用户自定义查询条件）查询，统计分析。

2. 基本操作

（1）语法关键字：

      show partitions，
      show tables，
      create table，
      load data (local) inpath，
      select * from，desc，alter，drop，
      limit，as，case when then end，union，
      like，group by，having，order by，sort by，cluster by...
1
2
3
4
5
6
7

（2）数据库操作

show databases;  \\ 显示数据库列表
use 数据库名;  \\使用数据库
set hive.cli.print.current.db=true; \\显示当前正在使用的数据库
1
2
3

（3）表操作

表的创建：

create table 表名(
    字段名 字段类型 comment , ... , 
    字段名 字段类型 comment) comment " ";

create table 表名(
    字段名 字段类型 comment , ... , 
    字段名 字段类型 comment) PARTITIONED BY (分区字段 字段类型)  指定分区
    row format delimited fields terminated by '\t'  自定义列分隔符
    STORED AS TEXTFILE;  \\自定义分隔符：hive表默认的解析方式----行列的分隔符\n
1
2
3
4
5
6
7
8
9

alter table 库名.表名 rename to 新库.新表; \\ 迁移表
alter table 表名 add columns (字段名 字段类型) ; \\增加一列
alter table 表名 drop column 字段名; \\删除一列
alter table表名 replace columns(字段名 字段类型); \\ 使用当前列替换掉原有的所有列
1
2
3
4

（4）数据操作

load data local inpath '数据所在路径' (overwrite) into table 库名.表名; \\导入本地数据
load data inpath 'user/hive/warehouser/...' (overwrite) into table ...;  \\导入HDSF数据
create table t1 as select * from t2;  \\查询导入
insert  (overwrite) into table t1 select * from t2;  \查询结果导入 
1
2
3
4

常用函数：https://www.iteblog.com/archives/2258.html#1

3. 数据抽样

（1）块抽样（不随机，按顺序返回数据，速度快）

create table t1 as select * from t2 tablesample(1000 rows);  \\ 指定行数
create table t1 as select * from t2 tablesample( 20 percent);  \\指定比例
create table t1 as select * from t2 tablesample( 1M);  \\指定数据大小
1
2
3

（2）分桶表抽样（随机，利用分桶表，随机分到多个桶，然后抽取指定的一个桶）

create table表as select * from 表 tablesample (bucket 1 out of 10 on rand());
1

（3）随机抽样（利用rand()函数抽取，rand()返回0~1之间的double值）

create table t1 as select * from t2 order by rand() limit 1000; 
 \\提供真正的随机抽样，但速度慢
create table t1 as select * from t2 sort by rand() limit 1000;
\\sort by 提供了单个 reducer 内的排序功能，速度有提升，但不保证整体随机
create table t1 as select * from t2 where rand()<0.002 distribute by rand() sort by rand() limit 10000; 
\\where 条件首先进行一次 map 端的优化，减少 reducer 需要处理的数据量，提高速度。distribute by 将数据随机分布，然后在每个 reducer 内进行随机排序，最终取10000条数据（如果数据量不足，可以提高 where 条件的 rand 过滤值），速度慢
create table t1 as select * from tw where rand() <0.002 cluster by rand() limit 10000;
\\cluster by 的功能是 distribute by 和 sort by 的功能相结合，distribute by rand() sort by rand() 进行了两次随机，cluster by rand() 仅一次随机，所以速度有提升。
1
2
3
4
5
6
7
8

（4）其他

1）随机结果添加分区：

 set hive.exec.dynamic.partition=true;
 set hive.exec.dynamic.partition.mode=nonstrict;
1
2

2）分桶：

分桶是相对分区进行更细粒度的划分。分桶将整个数据按照某列属性额hash值进行划分，如要按照name属性分为3个桶，就是对name属性值的hash值对3取摸，按照取模结果对数据分桶。
如何分桶：
第一，分桶之前要执行命令

hive.enforce.bucketiong=true;
1

第二，要使用关键字clustered by 指定分区依据的列名，还要指定分为多少桶，这里指定分为3桶。

create table t1 (id int,name string) clustered by(name) int 3 buckets 
row format delimited terminated by ‘\t’;
1
2

第三，与分区不同的是，分区依据的不是真实数据表文件中的列，而是我们指定的伪列，但是分桶是依据数据表中真实的列而不是伪列。所以在指定分区依据的列的时候要指定列的类型，因为在数据表文件中不存在这个列，相当于新建一个列。而分桶依据的是表中已经存在的列，这个列的数据类型显然是已知的，所以不需要指定列的类型。
第四，向桶中插入数据，

insert into table t1 select * from t2;
1

第五，查看分桶数据，n=1即返回第一桶数据。

select * from t1 tablesample(bucket n out of 3 on name);  
1

3）分组

row_number()：没有重复值的排序，可以利用它实现分页
dense_rank()：连续排序，两个第二名仍然跟着第三名
rank()：跳跃排序，两个第二名后跟着第四名

4. Hive性能及 SQL优化

（1）使用分区裁剪、列裁剪

尽量尽早地过滤数据，减少每个阶段的数据量,同时只选择需要使用到的字段。对于分区表要加分区，通过在条件中指定分区，来限制数据扫描范围，可以极大提高查询效率。
示例：

错误：
select * from t1 as a left join t2 as b on a.user_id = b.id 
where to_date(a.loan_time) >=’2018-02-01’ 
and a.etl_date =’2019-02-21’ and b.etl_date =’2019-02-21’; 
1
2
3
4

正确：
select a.loan_id,a.loan_time,b.user_name from
(select loan_id,loan_time from t1 
where etl_date =’2019-02-11’ and to_date(loan_time)>=’2018-02-01’)a 
left join 
(select id,user_nam from t2 where etl_date =’2019-02-11’)b on a.user_id = b.id;
1
2
3
4
5
6

（2）尽量避免使用count distinct

当数据量很大的情况下，一般count distinct使用先group by再count的方式替换，可以提高查询效率。
示例1：

select substr(loan_time,0,7),count(distinct utm_code) 
from t1 where etl_date =’2019-02-11’ 
group by substr(loan_time,0,7);
1
2
3

优化后：

select y_month,count(utm_code) from 
(select substr(loan_time,0,7) as y_month,utm_code from t1 
where etl_date =’2019-02-11’  group by substr(loan_time,0,7),utm_code
)a group by y_month;
1
2
3
4

示例2：

select utm_code,count(distinct user_id),count(distinct cont_code),
sum(cont_amount) from t2 where etl_date =’2019-02-01’;
1
2

优化后：

select utm_code,count(user_id),count(cont_code),sum(cont_amount) from(
select utm_code,null as user_id,cont_code,0 from t2 where etl_date =’2019-02-01’ 
group by utm_code,cont_code
Union all 
select utm_code, user_id,null as cont_code,0 from t2 where etl_date =’2019-02-01’ 
group by utm_code,user_id
Union all
select utm_code, null as user_id,null as cont_code,cont_amount from t2 where etl_date =’2019-02-01’ 
)tmp group by utm_code;
1
2
3
4
5
6
7
8
9

（3）JOIN优化

1）“小表放前”原则。

在编写带有 join 操作的代码语句时，应该将条目少的表/子查询放在 Join 操作符的左边，否则会引起磁盘和内存的大量消耗。

2）避免存在多对多的关联

执行SQL前调研一下是否存在多对多的关联，起码得保证有一个表或者结果集的关联键不重复。还有就是避免笛卡尔积，同理，如果某一个键的数据量非常大，也是很难完成Job的。

3）合理使用map join与SMB join

Map join主要适用于大表对小表（一张表的数据量很大，而另外一张表额数据量很少（小于1000行））和需要做不等值就操作。
set hive.auto.convert.join = true;自动识别大小表，把join变成合适的map join。
SMB join适用于大表对大表，设置如下。

set hive.auto.convert.sortmerge.join=true;
set hive.optimize.bucketmapjoin=true;
set hive.optimize.bucketmapjoin.sortedmerge=true;
1
2
3

（4）排序

order by排序只存在一个reduce，效率比较低。可采取sort by操作并结合distribute by做reduce分区键进行优化。
order by：全局排序，只使用一个reduce
sort by：对每一个reduce进行排序，局部排序
cluster by：对同一字段分桶并排序，不能和sort by连用
distribute by + sort by：分桶，保证同一字段值只存在一个结果文件中，结合sort by对每个reduce结果有序，distribute by必须在sort by之前

（5）优化in/exist语句

用semi join 替换in/exist,
示例：

select a.id, a.name from a where a.id in (select b.id from b);
select a.id, a.name from a where exists (select id from b where a.id = b.id);
1
2

应该转换成：

select a.id, a.name from a left semi join b on a.id = b.id;
1

（6）合理使用Union All

对同一张表的union all要比multi insert （即拆分成多个insert into语句）快得多。
对不同表的union all，如果union all 的部分个数大于2，或者每个union 的部分数据量大，则采用multi insert的效率会更高。

（7）并行执行Job

每个查询被hive转化成多个阶段，有些阶段关联性不大，则可以并行化执行，减少执行时间

set hive.exec.parallel=true;  \\打开任务并行执行
set hive.exec.parallel.thread.number=16; \\同一个sql允许最大并行度，默认为8。
1
2

PS：并行执行可以加快任务的执行速率，但不会减少其占用的资源。

（8）使用本地MR

如果在hive中运行的sql数据量比较小，则使用本地MR的效率会比提交到Hadoop集群中运行快很多。
开启本地模式：

set hive.exec.mode.local.auto =true;` \\设置本地MR的最大输入数据量，默认134217728，即128M：
set hive.exec.mode.local.auto.input.files.max = 50000000;  \\设置本地MR的最大输入文件个数，默认为4：
set hive.exec.mode.local.auto.input.files.max = 10;
1
2
3