首页  思维导图  详情

数据仓库-Hive

2021-04-12 08:55:39   0  举报





仅支持查看

AI智能生成

Hive笔记

hive

Hive的基本架构

hive基本原理图

作者其他创作

大纲/内容

数据仓库

概念

英文名：Data Warehouse

目的：构建面向分析的集成化数据环境，为企业提供决策支持

直接表现：存储数据，可以理解成面向分析的存储系统

特征

是面向主题的（Subject-Oriented ）

数据仓库是面向主题的,数据仓库通过一个个主题域将多个业务系统的数据加载到一起，为了 各个主题（如：用户、订单、商品等）进行分析而建，操作型数据库是为了支撑各种业务而 建立。

集成的（Integrated）

数据仓库会将不同源数据库中的数据汇总到一起,数据仓库中的综合数据不能从原有的数据库 系统直接得到。因此在数据进入数据仓库之前，必然要经过统一与整合，这一步是数据仓库 建设中最关键、最复杂的一步(ETL)，要统一源数据中所有矛盾之处，如字段的同名异义、异 名同义、单位不统一、字长不一致，等等。

非易失的（NonVolatile）

操作型数据库主要服务于日常的业务操作，使得数据库需要不断地对数据实时更新，以便迅 速获得当前最新数据，不至于影响正常的业务运作。 在数据仓库中只要保存过去的业务数据，不需要每一笔业务都实时更新数据仓库，而是根据 商业需要每隔一段时间把一批较新的数据导入数据仓库。数据仓库的数据反映的是一段相当 长的时间内历史数据的内容，是不同时点的数据库的集合，以及基于这些快照进行统计、综 合和重组的导出数据。数据仓库中的数据一般仅执行查询操作，很少会有删除和更新。但是 需定期加载和刷新数据。

时变的（Time-Variant ）

数据仓库包含各种粒度的历史数据。数据仓库中的数据可能与某个特定日期、星期、月份、 季度或者年份有关。数据仓库的目的是通过分析企业过去一段时间业务的经营状况，挖掘其 中隐藏的模式。虽然数据仓库的用户不能修改数据，但并不是说数据仓库的数据是永远不变 的。分析的结果只能反映过去的情况，当业务变化后，挖掘出的模式会失去时效性。因此数 据仓库的数据需要定时更新，以适应决策的需要。

数据库与数据仓库的区别

操作型处理，叫联机事务处理 OLTP（On-Line Transaction Processing，）

面向业务系统操作

分析型处理，叫联机分析处理 OLAP（On-Line Analytical Processing）

面向分析，支持决策

数仓的分层架构

源数据层（ODS）

清洗前的各类外围数据

数据仓库层（DW）

数据应该是一致的、准确的、干净的

数据应用层（DA或APP）

前端应用直接读取的数据源；根据报表、专题分析需求而计算 生成的数据。

数仓的元数据管理

作用

元数据（Meta Date），主要记录数据仓库中模型的定义、各层级间的映射关系、监控数据仓 库的数据状态及ETL的任务运行状态。一般会通过元数据资料库（Metadata Repository）来统 一地存储和管理元数据，其主要目的是使数据仓库的设计、部署、操作和管理能达成协同和 一致。

Hive

Hive 概念

Hive简介

什么是 Hive

Hive是基于Hadoop的一个数据仓库工具，可以将结构化的数据文件映射为一张数据库表，并 提供类SQL查询功能。 其本质是将SQL转换为MapReduce的任务进行运算，底层由HDFS来提供数据的存储。

为什么使用 Hive

<ol><li>采用类SQL语法去操作数据，提供快速开发的能力。</li><li>避免了去写MapReduce，减少开发人员的学习成本。</li><li>功能扩展很方便。</li></ol>

Hive架构

架构图

Hive 与 Hadoop 的关系

Hive与传统数据库对比

Hive 的安装

Hive 的交互方式

bin/hive

使用sql语句或者sql脚本进行交互

bin/hive -e "create database if not exists mytest;"

bin/hive -f /export/servers/hive.sql

Hive 操作

数据库操作

创建数据库

create database if not exists myhive; use myhive;

创建数据库并指定位置

create database if not exists myhive2 location '/user/hive/myhive2';

设置数据库键值对信息

create database if not exists myhvie3 with dbproperties('owner'='zhangsan','date'='20200912');

alter database myhive set dbproperties('owner'='wangwu','date'='20200912');

查看数据库键值对信息：describe database extended myhive;

查看数据库更多详细信息

desc database extended myhive2;

删除数据库

删除空数据库

drop database myhive4;

删除非空数据库(级联)

drop table myhive4 cascade;

数据库表操作

建表语法

create [external] table [if not exists] table_name ( col_name data_type [comment '字段描述信息'] col_name data_type [comment '字段描述信息']) [comment '表的描述信息'] [partitioned by (col_name data_type,...)] [clustered by (col_name,col_name,...)] [sorted by (col_name [asc|desc],...) into num_buckets buckets] [row format row_format] [storted as ....] [location '指定表的路径']

1. create table 创建一个指定名字的表。如果相同名字的表已经存在，则抛出异常；用户可以用 IF NOT EXISTS 选项来忽略这个异常。 2. external 可以让用户创建一个外部表，在建表的同时指定一个指向实际数据的路径 （LOCATION），Hive 创建内部表时，会将数据移动到数据仓库指向的路径；若创建外部 表，仅记录数据所在的路径，不对数据的位置做任何改变。在删除表的时候，内部表的 元数据和数据会被一起删除，而外部表只删除元数据，不删除数据。 3. comment 表示注释,默认不能使用中文 4. partitioned by 表示使用表分区,一个表可以拥有一个或者多个分区，每一个分区单独存在一个目录下 . 5. clustered by 对于每一个表分文件， Hive可以进一步组织成桶，也就是说桶是更为细粒 度的数据范围划分。Hive也是针对某一列进行桶的组织。 6. sorted by 指定排序字段和排序规则 7. row format 指定表文件字段分隔符 8. storted as指定表文件的存储格式, 常用格式:SEQUENCEFILE, TEXTFILE, RCFILE,如果文件 数据是纯文本，可以使用 STORED AS TEXTFILE。如果数据需要压缩，使用 storted as SEQUENCEFILE。 9. location 指定表文件的存储路径

内部表的操作

create table if not exists tab_name(id int, name string)

外部表的操作

create external table if not exists tab_name(id int, name string)

分区表的操作

create table tab_name(id string, name string) partitioned by (month string,day string) row format delimited fields terminated by '\t';

分区表综合练习

数据准备：

hdfs dfs -mkdir -p /scoredatas/month=201806 hdfs dfs -put score.csv /scoredatas/month=201806/

创建外部分区表，并指定文件数据存放目录

create external table score4(s_id string, c_id string,s_score int) partitioned by (month string) row format delimited fields terminated by '\t' location '/scoredatas';

进行表的修复 (建立表与数据文件之间的一个关系映射)

msck repair   table score4;

或者添加对应分区： alter table score4 add if not exists partition(month='201806') location '/scoredatas/month=201806';

分桶表操作

开启 Hive 的分桶功能

set hive.enforce.bucketing=true;

设置 Reduce 个数

set mapreduce.job.reduces=3;

创建分桶表

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';

创建普通表

create table course_common (c_id string,c_name string,t_id string) row format delimited fields terminated by '\t';

普通表中加载数据

load data local inpath '/export/servers/hivedatas/course.csv' into table course_common;

通过insert overwrite给桶表中加载数据

insert overwrite table course select * from course_common cluster by(c_id);

导入数据

导入Linux数据文件

load data local inpath '文件路径' into table tab_name;

导入HDFS数据文件

load data inpath 'HDFS文件路径' into table tab_name;

外部表修复

msck repair   table score4;

修改表结构

查看表的信息

分区

show partitions tab_name;

新增分区

alter table tab_name add partition(month='202008')

删除分区

alter table tab_name drop partition(month='202008')

模糊查看表

show tables like '*tab_n*';

查看表的创建语句

show create table tab_name;

表的结构信息（详细）

desc formatted tab_name;

简单查看表结构

desc tab_name;

表的扩展信息

desc extended tab_name;

重命名

alter table old_tab rename to new_tab;

增加/修改列信息

增加列

alter table tab_name add columns(sex string,birth string);

更新列

alter table mytable change name username string;

分区表加载数据

直接插入

insert into table score3 partition(month ='201807') values ('001','002','100');

通过查询插入数据

insert overwrite table tab_name partition(month='202008') select * from tab_name2;

通过load方式加载数据

load data local inpath '/linux数据文件地址' overwrite into table tab_name partition(month='202008');

导出数据

将查询结果导出指定目录

insert overwrite local directory '/export/servers/hivedatas/sort 'select * from score sort by s_score;

Hive 查询语法

语法

SELECT [ALL | DISTINCT] select_expr, select_expr, ... FROM table_reference [WHERE where_condition] [GROUPBY col_list [HAVING condition]] [CLUSTER BY col_list | [DISTRIBUTE BY col_list] [SORT BY | ORDERBY col_list]] [LIMIT number]

简单查询

select a.c_id, a.c_name from tab_name a;

常用函数

count(),max(),min(),sum(),avg()

limit 语句

select * from score limit3;

where子主题

比较运算符

A<=>B

如果A和B都为NULL，则返回TRUE，其他的和等号（=）操作符的结果一致，如果任一为NULL则结果为NULL

A RLIKE B, A REGEXP B

类型 B是一个正则表达式，如果A与其匹配，则返回TRUE；反之返回FALSE。匹配使用的是JDK中的正则表达式接口实现的，因为正则也依据其中的规则。例如，正则表达式必须和整个字符串A相匹配，而不是只需与其字符串匹配。

like 和 rlike

select * from score where s_id rlike '[1]'; #  like'%1%'

逻辑运算符

and,or,not

分组

GROUP BY

having语句，对group by 的结果再过滤

join语句

inner join

内连接

(也可简写为JOIN): 两个表中列匹配，则返回行

-- 例3.1 select * from t_user u, t_address a where u.id = a.user_id; -- 例3.2 select * from t_user u inner join t_address a where u.id = a.user_id; -- 例3.3 select * from t_user u inner join t_address a on u.id = a.user_id;

left join

左连接

right join

右连接

full join

全连接

只要其中一个表中存在匹配，就返回行

笛卡尔积

inner join 没有加on或者where条件

select * from t_user, t_address; 或 select * from t_user inner join t_address;

排序

Order By：全局排序，仅适合一个reduce（多个Reducer无法保证全局有序）

Sort By：局部排序，每个MapReduce内部进行排序，对全局结果集来说不是排序。

分区排序（DISTRIBUTE BY）

1.设置reduce的个数，将我们对应的s_id划分到对应的reduce当中去

set mapreduce.job.reduces=7;

2.通过distribute by 进行数据的分区

insert overwrite local directory '/export/servers/hivedatas/sort' select * from score distribute by s_id sort by s_score;

CLUSTER BY

等价效果： select * from score cluster by s_id; select * from score distribute by s_id sort by s_id;

子主题

Hive Shell参数

Hive命令行

语法

bin/hive [-hiveconf x=y]* [<-i filename>]* [<-f filename>|<-e query-string>] [-S]

1、 -i 从文件初始化HQL。 2、-e从命令行执行指定的HQL。 3、-f 执行HQL脚本 4、 -v 输出执行的HQL语句到控制台5、 -p connect to Hive Server on port number6、 -hiveconf x=y Use this to set hive/hadoop configuration variables. 设置hive运行时候的参数配置

Hive参数配置方式

配置文件

用户自定义配置文件：$HIVE_CONF_DIR/hive-site.xml 默认配置文件：$HIVE_CONF_DIR/hive-default.xml 用户自定义配置会覆盖默认配置。

命令行参数

bin/hive -hiveconf hive.root.logger=INFO,console

参数声明

set mapred.reduce.tasks=100;

Hive 函数

内置函数

1. 查看系统自带的函数

show functions;

2. 显示自带的函数的用法

desc function upper;

3. 详细显示自带的函数的用法

desc function extended upper;

4:常用内置函数

自定义函数(包含三种)

1.UDF（User-Defined-Function）

一进一出

2.UDAF（User-Defined Aggregation Function）

聚集函数；多进一出（count，max，min）

3.UDTF（User-Defined Table-Generating Functions）

一进多出（lateral，view，explore）

删除临时函数

drop temporary function myhive.my_upper;

自定函数开发实例（Java）

上传自定义函数java jar包到hive lib目录下

在 hive客户端执行:add jar /自定义jar包位置;

在hive客户端创建自定义函数名：create temporary function my_upper as'cn.itcast.udf.ItcastUDF';

使用自定义函数：select my_upper('abc');

Hive的数据压缩

子主题

压缩配置参数

输入端压缩

io.compression.codecs （在core-site.xml中配置）

org.apache.hadoop.io.compress.DefaultCodec, org.apache.hadoop.io.compress.GzipCodec, org.apache.hadoop.io.compress.BZip2Codec, org.apache.hadoop.io.compress.Lz4Codec

map输出端

mapreduce.map.output.compress

false

mapreduce.map.output.compress.codec

org.apache.hadoop.io.compress.DefaultCodec

reduce输出端

mapreduce.output.fileoutputformat.compress

false

mapreduce.output.fileoutputformat.compress.codec

org.apache.hadoop.io.compress. DefaultCodec

mapreduce.output.fileoutputformat.compress.type

RECORD

开启Map输出阶段压缩

1.开启hive中间传输数据压缩功能

set hive.exec.compress.intermediate=true;

2.开启mapreduce中map输出压缩功能

set mapreduce.map.output.compress=true;

3.设置mapreduce中map输出数据的压缩方式

set mapreduce.map.output.compress.codec= org.apache.hadoop.io.compress.SnappyCodec;

开启Reduce输出阶段压缩

1.开启hive最终输出数据压缩功能

set hive.exec.compress.output=true;

2.开启mapreduce最终输出数据压缩

set mapreduce.output.fileoutputformat.compress=true;

3.设置mapreduce最终数据输出压缩方式

set mapreduce.output.fileoutputformat.compress.codec=org.apache.hadoop.io.compress.SnappyCodec

4.设置mapreduce最终数据输出压缩为块压缩

set mapreduce.output.fileoutputformat.compress.type=BLOCK;

Hive的数据存储格式

列式存储和行式存储

行存储的特点

查询满足条件的一整行数据的时候，列存储则需要去每个聚集的字段找到对应的每个列的值，行存储只需要找到其中一个值，其余的值都在相邻地方，所以此时行存储查询的速度更快。

列存储的特点

因为每个字段的数据聚集存储，在查询只需要少数几个字段的时候，能大大减少读取的数据量；每个字段的数据类型一定是相同的，列式存储可以针对性的设计更好的设计压缩算法。

常用的数据存储格式

TEXTFILE（行式存储）

默认格式，数据不做压缩，磁盘开销大，数据解析开销大。可结合Gzip、Bzip2使用.

SEQUENCEFILE(行式存储)

ORC（列式存储）

Orc (Optimized Row Columnar)是hive 0.11版里引入的新的存储格式。 可以看到每个Orc文件由1个或多个stripe组成，每个stripe250MB大小，每个Stripe里有三部分组成，分别是Index Data,Row Data,Stripe Footer：

PARQUET(列式存储)

Parquet是面向分析型业务的列式存储格式; Parquet文件是以二进制方式存储的，所以是不可以直接读取的，文件中包括该文件的数据和 元数据，因此Parquet格式文件是自解析的。

配置属性

parquet.compression parquet.block.size parquet.page.size dfs.blocksize mapred.max.split.size

支持类型：uncompressed", "snappy", "gzip", "lzo"。 Snappy压缩具有更好的性能，Gzip压缩具有更好的压缩比。

建表声明

CREATE TABLE parquet_test ( id int, str string, mp MAP<STRING,STRING>, lst ARRAY<STRING>, strct STRUCT<A:STRING,B:STRING>) PARTITIONED BY (part string) STORED AS PARQUET;

文件存储格式与数据压缩结合

压缩比和查询速度对比

TextFile

createtable log_text (track_time string,url string,session_id string,referer string,ip string,end_user_id string,city_id string)ROW FORMAT DELIMITED FIELDS TERMINATED BY'\t'STORED AS TEXTFILE ;

load data local inpath '/export/servers/hivedatas/log.data'intotable log_text ;

查看表中数据大小: dfs -du -h /user/hive/warehouse/myhive.db/log_text;

ORC

createtable log_orc(track_time string,url string,session_id string,referer string,ip string,end_user_id string,city_id string)ROW FORMAT DELIMITED FIELDS TERMINATED BY'\t'STORED AS orc ;

insertintotable log_orc select * from log_text ;

dfs -du -h /user/hive/warehouse/myhive.db/log_orc;

ParRuet

createtable log_parquet(track_time string,url string,session_id string,referer string,ip string,end_user_id string,city_id string)ROW FORMAT DELIMITED FIELDS TERMINATED BY'\t'STORED AS PARQUET ;

insertintotable log_parquet select * from log_text ;

dfs -du -h /user/hive/warehouse/myhive.db/log_parquet;

文件压缩总结：ORC > Parquet > textFile

存储文件的查询速度总结：ORC > Parquet > TextFile （parquet与textfile差不多）

ORC存储指定压缩方式

ORC存储方式的压缩

创建一个非压缩的的ORC存储方式

create table log_orc_snappy(track_time string,url string,session_id string,referer string,ip string,end_user_id string,city_id string)ROW FORMAT DELIMITED FIELDS TERMINATED BY'\t'STORED AS orc tblproperties ("orc.compress"="SNAPPY");

insert into table log_orc_snappy select * from log_text ;

dfs -du -h /user/hive/warehouse/myhive.db/log_orc_snappy ;dfs -du -h /user/hive/warehouse/myhive.db/log_orc_snappy ;

创建一个SNAPPY压缩的ORC存储方式

insert into table log_orc_snappy select * from log_text ;

dfs -du -h /user/hive/warehouse/myhive.db/log_orc_snappy ;

存储方式和压缩总结

在实际的项目开发当中，hive表的数据存储格式一般选择：orc或parquet。压缩方式一般选择snappy。(使用压缩后可大量节省空间)

Hive调优

1.Fetch抓取

把hive.fetch.task.conversion设置成none，然后执行查询语句，都会执行mapreduce程序。

set hive.fetch.task.conversion=none;

把hive.fetch.task.conversion设置成more，然后执行查询语句，如下查询方式都不会执行mapreduce程序。

set hive.fetch.task.conversion=more;

2.本地模式

大多数的Hadoop Job是需要Hadoop提供的完整的可扩展性来处理大数据集的。不过，有时Hive的输入数据量是非常小的。在这种情况下，为查询触发执行任务时消耗可能会比实际job 的执行时间要多的多。对于大多数这种情况，Hive可以通过本地模式在单台机器上处理所有的任务。对于小数据集，执行时间可以明显被缩短。

开启：set hive.exec.mode.local.auto=true;

关闭：set hive.exec.mode.local.auto=false;

3.MapJoin

如果不指定MapJoin或者不符合MapJoin的条件，那么Hive解析器会在Reduce阶段完成join,容易发生数据倾斜。可以用MapJoin把小表全部加载到内存在map端进行join，避免reducer处理。

开启MapJoin参数设置

set hive.auto.convert.join = true;

set hive.mapjoin.smalltable.filesize=25123456;

4.Group By

开启Map端聚合参数设置

（1）是否在Map端进行聚合，默认为True

set hive.map.aggr = true;

（2）在Map端进行聚合操作的条目数目

set hive.groupby.mapaggr.checkinterval = 100000;

（3）有数据倾斜的时候进行负载均衡（默认是false）

 set hive.groupby.skewindata = true;

5.Count(distinct)

数据量小的时候无所谓，数据量大的情况下，由于COUNT DISTINCT操作需要用一个ReduceTask来完成，这一个Reduce需要处理的数据量太大，就会导致整个Job很难完成，一般COUNT DISTINCT使用先GROUP BY再COUNT的方式替换：

select count(distinct s_id) from score; select count(s_id) from (select s_id from score group by s_id) a;

6.笛卡尔积

尽量避免笛卡尔积，即避免join的时候不加on条件，或者无效的on条件，Hive只能使用1个reducer来完成笛卡尔积。

7.动态分区调整

开启动态分区参数设置

（1）开启动态分区功能（默认true，开启）

set hive.exec.dynamic.partition=true;

（2）设置为非严格模式 （动态分区的模式，默认strict，表示必须指定至少一个分区为静态分区，nonstrict模式表示允许所有的分区字段都可以使用动态分区。）

set hive.exec.dynamic.partition.mode=nonstrict;

（3）在所有执行MR的节点上，最大一共可以创建多少个动态分区。

set hive.exec.max.dynamic.partitions=1000;

（4）在每个执行MR的节点上，最大可以创建多少个动态分区。该参数需要根据实际的数据来设定。

<div>set hive.exec.max.dynamic.partitions.pernode=100</div>

（5）整个MR Job中，最大可以创建多少个HDFS文件。

set hive.exec.max.created.files=100000;

（6）当有空分区生成时，是否抛出异常。一般不需要设置。

set hive.error.on.empty.partition=false;

案例操作

（1）准备数据原表

（2）创建目标分区表

createtable ori_partitioned_target(id bigint, timebigint, uid string, keyword string, url_rank int, click_num int, click_url string) PARTITIONED BY (p_time STRING) row format delimited fields terminated by'\t'

（3）向目标分区表加载数据

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;

（4）查看分区

show partitions ori_partitioned_target;

8.并行执行

Hive会将一个查询转化成一个或者多个阶段。这样的阶段可以是MapReduce阶段、抽样阶段、合并阶段、limit阶段。或者Hive执行过程中可能需要的其他阶段。默认情况下，Hive一次只会执行一个阶段。不过，某个特定的job可能包含众多的阶段，而这些阶段可能并非完全互相依赖的，也就是说有些阶段是可以并行执行的，这样可能使得整个job的执行时间缩短。不过，如果有更多的阶段可以并行执行，那么job可能就越快完成。

set hive.exec.parallel = true;

9.严格模式

Hive提供了一个严格模式，可以防止用户执行那些可能意向不到的不好的影响的查询。 通过设置属性hive.mapred.mode值为默认是非严格模式nonstrict 。 开启严格模式需要修改hive.mapred.mode值为strict，开启严格模式可以禁止3种类型的查询。

对于分区表，在where语句中必须含有分区字段作为过滤条件来限制范围，否则不允许执行

对于使用了order by语句的查询，要求必须使用limit语句

限制笛卡尔积的查询

10.JVM重用

JVM重用是Hadoop调优参数的内容，其对Hive的性能具有非常大的影响，特别是对于很难避免小文件的场景或task特别多的场景，这类场景大多数执行时间都很短。Hadoop的默认配置通常是使用派生JVM来执行map和Reduce任务的。这时JVM的启动过程可能会造成相当大的开销，尤其是执行的job包含有成百上千task任务的情况。JVM重用可以使得JVM实例在同一个job中重新使用N次。N的值可以在Hadoop的mapred-site.xml文件中进行配置。通常在10-20之间，具体多少需要根据具体业务场景测试得出。

set mapred.job.reuse.jvm.num.tasks=10;

缺点：

开启JVM重用将一直占用使用到的task插槽，以便进行重用，直到任务完成后才能释放。如果某个“不平衡的”job中有某几个reduce task执行的时间要比其他Reduce task消耗的时间多的多的话，那么保留的插槽就会一直空闲着却无法被其他的job使用，直到所有的task都结束了才会释放。

11.推测执行

设置开启推测执行参数：

set mapred.map.tasks.speculative.execution=true； set mapred.reduce.tasks.speculative.execution=true； set hive.mapred.reduce.tasks.speculative.execution=true;

在分布式集群环境下，因为程序Bug（包括Hadoop本身的bug），负载不均衡或者资源分布不均等原因，会造成同一个作业的多个任务之间运行速度不一致，有些任务的运行速度可能明显慢于其他任务（比如一个作业的某个任务进度只有50%，而其他所有任务已经运行完毕），则这些任务会拖慢作业的整体执行进度。为了避免这种情况发生，Hadoop采用了推测执行（Speculative Execution）机制它根据一定的法则推测出“拖后腿”的任务，并为这样的任务启动一个备份任务，让该任务与原始任务同时处理同一份数据，并最终选用最先成功运行完成任务的计算结果作为最终结果。

Hive的explain

语法

EXPLAIN [EXTENDED|DEPENDENCY|AUTHORIZATION] query

例子：explain select * from score cluster by s_id;