Hive学习笔记，
1. Hive简介
Hive是基于Hadoop的一个数据仓库工具,可以将结构化的数据文件映射为一张数据库表,并提供类SQL的查询功能,其本质是将SQL转换成MapReduce的任务进行运算,底层由HDFS来提供数据的存储.
2. 特点
- 可扩展:Hive可自由的扩展集群的规模,一般情况下不需要重启服务
- 延展性:Hive支持用户根据需求来自定义函数
- 容错性:节点出现问题时SQL仍可完成执行(依赖于Hadoop的特性)
3. Hive的元数据
- Hive的元数据(Meta Data)主要记录数据仓库中模型的定义,各层级间的映射关系,监控数据仓库的数据状态及ETL的任务运行状态
- Hive中元数据默认存储在自带的Derby内存数据库中,但是这样对于元数据的持久化和数据的安全性都没有保障,所以一般将Hive的元数据存储在MySQL/Oracle中
- 元数据可分为:
- 技术元数据:包含了数据间的各种映射关系,主要提供给开发和管理数据仓库的IT人员使用
- 业务元数据:提供给管理层和业务分析人员使用
4. Hive的数据存储
-
Hive中所有的数据都存储在 HDFS 中，没有专门的数据存储格式（可支持Text，SequenceFile，ParquetFile，ORC格式等）
-
SequenceFile是hadoop中的一种文件格式:文件内容是以序列化的K-V对象来组织的
-
只需要在创建表的时候告诉Hive数据中的列分隔符和行分隔符，Hive就可以解析数据
5. Hive的基本操作
5.1 数据库相关操作
以下所有的数据库名称统一使用DBN(DataBaseName)代替,数据库表名称统一使用TB(TableName)代替
- 创建数据库:create database if not exists DBN
- 使用数据库:use DBN
- 创建数据库并指定HDFS的存储路径:create database if not exists DBN location ‘/myhive’
- 修改数据库(数据库的元数据信息一旦创建不可更改,包括数据库的名称和所在的位置):alert database DBN set
- 查看指定数据库的详细信息:desc database DBN 或者 desc database extended DBN
- 删除一个空的数据库:drop database DBN
5.2 数据库表相关操作
- 创建数据库表:
CREATE [EXTERNAL] TABLE [IF NOT EXISTS] table_name
[(col_name data_type [COMMENT col_comment], ...)]
[COMMENT table_comment]
[PARTITIONED BY (col_name data_type [COMMENT col_comment], ...)]
[CLUSTERED BY (col_name, col_name, ...)
[SORTED BY (col_name [ASC|DESC], ...)] INTO num_buckets BUCKETS]
[ROW FORMAT row_format]
[STORED AS file_format]
[LOCATION hdfs_path]
- 创建数据库表并指定字段之间的分隔符:
create table if not exists TB(id int ,name string) row format delimited fields terminated by '\t' stored as textfile location '/user/stu1';
- 根据查询结果创建表(复制内容):
create table TB2 as select * from TB1;
- 根据已经存在的表的结构创建新表(复制表结构):
create table TB4 like TB3;
- 查询表的类型:
desc formatted TB;
5.3 内部表,外部表
- 内部表:
- 数据私有化
- 在记录数据之间的关系的同时将数据移动至Hive指定的目录
- 删除内部表的时候数据会被一起删除
- 清空管理表:truncate table TB;
- 外部表:
- 关键字:external
- 数据公有化
- 只记录数据之间的关系,不实际移动数据
- 删除外部表的时候数据不会被删除
#从本地文件系统向表中加载数据
load data local inpath '/export/servers/hivedatas/student.csv' into table student;
#加载数据并覆盖已有数据
load data local inpath '/export/servers/hivedatas/student.csv' overwrite into table student;
5.4 分区表,分桶表
- 分区表:
- 关键字:partitioned by(名称 类型)
- 分完后的是文件夹,便于局部查询
create table score(s_id string,c_id string, s_score int) partitioned by (month string) row format delimited fields terminated by '\t';
- 分桶表:
- 关键字:clustered by(字段) into 3 buckets
- 分的是文件,即将一个文件拆分为多个小文件
create table course (c_id string,c_name string,t_id string) clustered by(c_id) into 3 buckets row format delimited fields terminated by '\t';
- 往表中插入数据时:
#指定从本地文件中加载数据到表中(保留原来的数据)
load data local inpath '/export/servers/hivedatas/score.csv' overwrite into table score partition(month='201806');
#指定从HDFS中加载数据到表中(删除原来的数据)
load data inpath '/hivedatas/score.csv' overwrite into table score partition(month='201806');
6. Hive的查询语法
具体参考SQL的查询语法,在此仅列出需要注意和有区别的地方
- order by:全局排序,故只能有一个reducer
- sort by:局部排序
- distribute by(字段):根据指定的字段将数据分到不同的reducer,分发算法采用的是hash散列
- cluster by(字段):除了具有distribute by的功能外,还会对该字段进行排序
- 若分桶和sort字段是同一个时,cluster by = distribute by + sort by
- 强烈建议查询的时候在查询语句最后加上
limit n以提高查询效率 - Hive中的
like和rlike在使用的时候需要注意区分 - group by后面的字段和select后面的字段要保持一致
- Hive支持通常的SQL join语句,但是只支持等值连接,不支持非等值连接
7. Hive的函数
- 内置函数的使用参考官方文档
- 用户可自定义的三种函数:
- UDF:一进一出
- UDAF:聚集函数,多进一出
- UDTF:一进多出
- 自定义函数的步骤:
- 继承UDF类
- 重载evaluate方法
- 定义函数功能
- 打成jar包上传到hive的lib目录下
- 设置函数与自定义函数的关联
- 执行自定义函数
8. Hive的数据压缩
-
目的是为了减轻MapReduce执行任务时的IO负担和网络带宽,提高效率
-
Hive支持的数据压缩有:gzip,bzip2,LZO,LZ4等
-
通过自己动手编译CDH版的Hadoop,可使Hive支持谷歌的Snappy数据压缩算法,性能炸裂
-
注意:Hive依赖于Hadoop,故Hadoop支持的数据压缩算法Hive都支持
-
Hive的数据压缩分为Map阶段的数据压缩和Reduce阶段的数据压缩,均可单独控制
9. Hive的数据存储格式
| 存储格式 | 存储类型 |
|---|---|
| TextFile | 行式存储 |
| SequenceFile | 行式存储 |
| Parquet | 列式存储 |
| ORC | 列式存储 |
10. Hive的调优
方案一：fetch抓取
-
这个属性配置有三个取值 more minimal none 如果配置成none，所有的都要走mr程序
set hive.fetch.task.conversion=more 表示全局查找，字段查找，limit查找都不走mr -
hive的本地模式：
set hive.exec.mode.local.auto=true 开启本地模式，解决多个小文件输入的时候，分配资源时间超过数据的计算时间 -
set hive.exec.mode.local.auto.inputbytes.max=51234560; 设置输入的数据临界值，如果小于这值都认为是小任务模式，启动本地模式来执行
-
set hive.exec.mode.local.auto.input.files.max=10; 设置输入文件个数的临界值，如果小于这个数量，那么也认为是小任务模式
方案二：hive表的优化
- 去重的优化：
select count(distinct s_id) from score;这种写法所有的去重数据都会在一个reduce当中去，造成数据处理比较慢
select count(1) from (select s_id from score group by s_id) bysid; 这种写法，使用了一个嵌套子查询，先对数据进行group by去重，然后再进行统计
尽量避免大sql，可以将一个很大的sql拆成多段，分步的去执行
- 大表join大表的优化：
空key的过滤
不过滤：
INSERT OVERWRITE TABLE jointable SELECT a.* FROM nullidtable a JOIN ori b ON a.id = b.id;
结果：No rows affected (152.135 seconds)
过滤：过滤掉我们所有的为null的id，使得输入数据量变少
INSERT OVERWRITE TABLE jointable SELECT a.* FROM (SELECT * FROM nullidtable WHERE id IS NOT NULL ) a JOIN ori b ON a.id = b.id;
结果：No rows affected (141.585 seconds)
空key的转换:
如果规定这些空key过滤不调，那么我们可以对空key进行转换
SELECT a.* FROM nullidtable a LEFT JOIN ori b ON CASE WHEN a.id IS NULL THEN ‘hive’ ELSE a.id END = b.id;
如果空key比较多，那么就会将大量的空key转换成 hive，那么就会遇到一个问题，数据倾斜
- 数据倾斜的表现形式：有一个reduce处理的数据量远远比其他reduce处理的数据量要大，造成其他的reduce数据都处理完了，这个还没处理完
怎么发现的数据倾斜，如何出现的数据倾斜，怎么解决的数据倾斜?
空key的打散:
SELECT a.* FROM nullidtable a LEFT JOIN ori b ON CASE WHEN a.id IS NULL THEN concat(‘hive’, rand()) ELSE a.id END = b.id;
通过将空key打散成不同的随机字符串，就可以解决hive的数据倾斜的问题
方案三：map端的join
hive已经开启了自动的map端的join功能，不管是大表join小表，还是小表join大表，都会将小表加载到内存当中来
- 首先第一步：启动一个local的task，寻找哪个表的数据是小表数据
hive的group by优化：能在map端聚合的数据，就尽量在map端进行聚合
多加一层mr的程序，让数据实现均衡的负载，避免数据的倾斜
- count(distinct)的优化：
这种写法效率低下：SELECT count(DISTINCT id) FROM bigtable;
可以转换成这种写法：SELECT count(id) FROM (SELECT id FROM bigtable GROUP BY id) a;
- 笛卡尔积：任何时候都要避免笛卡尔积，避免无效的on条件
select from A left join B – on A.id = B.id
- 使用分区裁剪，列裁剪：
分区裁剪：如果是分区表，那么查询的时候，尽量带上分区条件
列裁剪：尽量避免使用select * ,需要查询哪些列，就选择哪些列
- 动态分区调整：
分区表的数据加载两种方式：
load data inpath ‘/export/xxx’ into table xxx partition(month = ‘xxx’)
insert overwrite table xxx partition (month = ‘xxx’) select xxx
- 使用动态分区动态的添加数据
如果要使用动态分区添加数据，最后一个字段一定要是分区字段
INSERT overwrite TABLE ori_partitioned_target PARTITION (p_time) SELECT id, time, uid, keyword, url_rank, click_num, click_url, p_time FROM ori_partitioned;
- 数据的倾斜：
主要就是合理的控制map个数以及reduce个数
第一个问题：maptask的个数怎么定的？？？与我们文件的block块相关，默认一个block块就是对应一个maptask
第二个问题：reduceTask的个数怎么定的？？？是我们自己手动设置的，爱设几个设几个，没人管你
第三个问题：是不是maptask的个数越多越好：不一定：有时候有些小文件，都要启动一个maptask，分配资源的时间超过了数据处理的时间
减少mapTask的个数：设置map端的小文件合并：使用combineHiveInputFormat来实现对我们小文件的合并，减少maptask的个数 或者使用本地模式也可以解决小文件的问题
增加maptask的个数：我们可以多设置几个reduce，然后使用distribte by将数据打散
set mapreduce.job.reduces =10;
create table a_1 as select * from a distribute by rand(123);
- 第四个问题：控制reduceTask的个数：
reduce个数设置方法：
（1）每个Reduce处理的数据量默认是256MB
hive.exec.reducers.bytes.per.reducer=256123456
（2）每个任务最大的reduce数，默认为1009
hive.exec.reducers.max=1009
（3）计算reducer数的公式
N=min(参数2，总输入数据量/参数1)
直接凭感觉设置reduce的个数：
set mapreduce.job.reduces = 15;
方案四:使用Explain(执行计划)
查看执行计划：
explain extended select * from course;
方案五:并行执行
有时候有些sql之间是不相关的，可以并行的一起执行，那么就可以用并行执行
方案六:严格模式
如果开启hive的严格模式，有以下三个限制
1、分区表需要带上分区字段
2、order by 必须使用limit
3、笛卡尔积不能执行
方案七:JVM的重用
container的里面的任务执行完成之后，不要马上释放资源，留着资源给下一个任务执行
方案八:推测执行
maptask的推测执行以及reducetask的推测执行：
一般都直接关闭maptask以及reducetask的推测执行
set mapreduce.map.speculative=false;关闭map端的推测执行
set mapreduce.reduce.speculative=false; 关闭reduce端的推测执行
方案九:压缩与存储
压缩：snappy 源始数据存储：TextFile
处理之后的数据存储：ORC PARQUET