数据仓库 Hive（2）

6、经典案例
- 6.1、WordCount
- - 指标
  - 处理方案
- 6.2、天气系统
- - 指标
  - 样例数据
  - 处理方案
- 6.3、好友推荐
- - 指标
  - 样例数据
  - 处理方案
7、数据倾斜
- 7.1、定义
- 7.2、原因
8、Hive文件存储格式
- 8.1、文件存储方式
- 8.2、文件存储格式
- 8.3、文件压缩练习
9、Hive参数
- 9.1、设置参数的三种方式
- 9.2、常用的hive设置
10、Hive企业优化
- 10.1、Fetch
- 10.2、本地模式
- 10.3、并行执行
- 10.4、严格模式
- 10.5、JVM重用
- 10.6、表的优化（小表与大表）
- 10.6、表的优化（大表与大表）
- 10.7、mapside聚合
- 10.8、Count(Distinct)
- 10.9、防止笛卡尔积

6、经典案例

6.1、WordCount

指标

计算每个单词出现的次数？

处理方案

-- 表结构
create table t_wordcount(
line string
)
row format delimited lines terminated by '\n';-- 加载数据
load data inpath '/yjx/harry.txt' into table t_wordcount;-- 处理方案
select regexp_replace(line,"\[\^0-9a-zA-Z\`\'\\-\\s\]","") from t_wordcount;select split(regexp_replace(line,"\[\^0-9a-zA-Z\`\'\\-\\s\]",""),"\\s+") from t_wordcount;select explode(split(regexp_replace(line,"\[\^0-9a-zA-Z\`\'\\-\\s\]",""),"\\s+")) word from t_wordcount;select word,count(word) from t_wordcount, lateral view explode(split(regexp_replace(line,"\[\^0-9a-zA-Z\`\'\\-\\s\]",""),"\\s+")) wordtable as  word group by word;

6.2、天气系统

指标

每个市每天的最高温度和最低温度和平均温度
每天每个省最热的城市是那一个？
查询出每个省份每个月最高的三个温度和日期，最低的三个温度和日期？
综合每月数据查询出那个城市的上报日期与创建日期延迟最高？
查询出 7月份全国晴天最多的省份

样例数据

1199,广东,广州市,440100,多云,25,西南,≤3,97,3/6/2020 01:52:18,3/6/2020 02:00:01

处理方案

--创建表
create table t_weather(
id int,
province string,
city string,
adcode int,
weather string,
temperature int,
winddirection string,
windpower string,
humidity int,
reporttime string,
createtime string
)
row format delimited fields terminated by ','
lines terminated by '\n';
-- 载入数据
load data inpath '/yjx/weather.txt' into table t_weather;--解决方案 每个市每天的最高温度和最低温度和平均温度
select * from t_weather limit 0,10;select UNIX_TIMESTAMP(reporttime,'dd/MM/yyyy HH:mm:ss') from t_weather limit 0,10;select from_unixtime(UNIX_TIMESTAMP(reporttime,'dd/MM/yyyy HH:mm:ss'),'yyyyMMdd') from t_weather limit 0,10;select adcode , from_unixtime(UNIX_TIMESTAMP(reporttime,'dd/MM/yyyy HH:mm:ss'),'yyyyMMdd'),max(temperature),min(temperature),avg(temperature) from t_weather group by adcode , from_unixtime(UNIX_TIMESTAMP(reporttime,'dd/MM/yyyy HH:mm:ss'),'yyyyMMdd') ;-- 解决方案 每天每个省最热的城市是那一个？
select province,city,from_unixtime(UNIX_TIMESTAMP(reporttime,'dd/MM/yyyy HH:mm:ss'),'yyyyMMdd') ymd,temperature,max(temperature) over(partition by province,from_unixtime(UNIX_TIMESTAMP(reporttime,'dd/MM/yyyy HH:mm:ss'),'yyyyMMdd') order by temperature ) mt from t_weather ;select province,city,ymd,temperature from (select province,city,from_unixtime(UNIX_TIMESTAMP(reporttime,'dd/MM/yyyy HH:mm:ss'),'yyyyMMdd') ymd,temperature,row_number() over(partition by province,from_unixtime(UNIX_TIMESTAMP(reporttime,'dd/MM/yyyy HH:mm:ss'),'yyyyMMdd') order by temperature desc ) rn from t_weather ) hottable where hottable.rn <= 1;

6.3、好友推荐

指标

字段：间接好友直接好友简介好友推荐度

样例数据

何玉洁,郭燕芳-蔡雯蔚-黄龙剑-李鸿涛

处理方案

-- 创建表
create table t_friend(
name string,
friends array<string>
)
row format delimited fields terminated by ','
collection items terminated by '-'
lines terminated by '\n';-- 载入数据
load data inpath '/yjx/friend.txt' into table t_friend;-- 解决方案-- 1求出直接好友的数据 name:name -1
select concat(name,":",friend) fs ,-1 num from t_friend , lateral view explode(friends) ft as friend;-------------------------------------- 间接好友推荐过程 start
select * from (select explode(friends) ef from t_friend) f1 ;select f1.name,f1.fd from (select name,fd from t_friend lateral view explode(friends) f1 as fd) f1 ;select f1.name,f1.fd ,f2.name,f2.fd from (select name,fd from t_friend lateral view explode(friends) f1 as fd) f1 ,(select name,fd from t_friend lateral view explode(friends) f1 as fd) f2 ;select f1.name,f1.fd ,f2.name,f2.fd from (select name,fd from t_friend lateral view explode(friends) f1 as fd) f1 ,(select name,fd from t_friend lateral view explode(friends) f1 as fd) f2 where f1.name = f2.name AND f1.fd <> f2.fd;select f1.name,f1.fd ,f2.name,f2.fd from (select name,fd from t_friend lateral view explode(friends) f1 as fd) f1 ,(select name,fd from t_friend lateral view explode(friends) f1 as fd) f2 where f1.name = f2.name and f1.fd <> f2.fd;
-------------------------------------- 间接好友推荐过程 end-- 2求出间接好友的数据  name:name 0
select concat(f1.fd,":",f2.fd) fs,0 num from (select name,fd from t_friend lateral view explode(friends) f1 as fd) f1 ,(select name,fd from t_friend lateral view explode(friends) f1 as fd) f2 where f1.name = f2.name and f1.fd <> f2.fd;-- 3合并直接好友和间接好友
select ww.fs,ww.num from (select concat(name,":",friend) fs ,-1 num from t_friend , lateral view explode(friends) ft as friend union all (select concat(f1.fd,":",f2.fd) fs,0 num from (select name,fd from t_friend lateral view explode(friends) f1 as fd) f1 ,(select name,fd from t_friend lateral view explode(friends) f1 as fd) f2 where f1.name = f2.name and f1.fd <> f2.fd))  ww;-- 4按照名字组合进行分组，求出推荐度
select ww.fs,count(ww.num) from (select concat(name,":",friend) fs ,-1 num from t_friend , lateral view explode(friends) ft as friend union all (select concat(f1.fd,":",f2.fd) fs,0 num from (select name,fd from t_friend lateral view explode(friends) f1 as fd) f1 ,(select name,fd from t_friend lateral view explode(friends) f1 as fd) f2 where f1.name = f2.name and f1.fd <> f2.fd))  ww group by ww.fs having sum(ww.num) = 0;-- 5将数据写出到HDFS
insert overwrite directory '/yjx/result/friend' ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
select ww.fs,count(ww.num) from (select concat(name,":",friend) fs ,-1 num from t_friend , lateral view explode(friends) ft as friend union all (select concat(f1.fd,":",f2.fd) fs,0 num from (select name,fd from t_friend lateral view explode(friends) f1 as fd) f1 ,(select name,fd from t_friend lateral view explode(friends) f1 as fd) f2 where f1.name = f2.name and f1.fd <> f2.fd))  ww group by ww.fs having sum(ww.num) = 0;-- 优化策略（直接好友）
select concat(name,":",friend) fs ,-1 num from t_friend , lateral view explode(friends) ft as friend where name < friend;
-- 优化策略（间接好友）
select concat(f1.fd,":",f2.fd) fs,0 num from (select name,fd from t_friend lateral view explode(friends) f1 as fd) f1 ,(select name,fd from t_friend lateral view explode(friends) f1 as fd) f2 where f1.name = f2.name and f1.fd < f2.fd;
-- 最终
insert overwrite directory '/yjx/result/friend' ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
select ww.fs,count(ww.num) from (select concat(name,":",friend) fs ,-1 num from t_friend , lateral view explode(friends) ft as friend where name < friend union all (select concat(f1.fd,":",f2.fd) fs,0 num from (select name,fd from t_friend lateral view explode(friends) f1 as fd) f1 ,(select name,fd from t_friend lateral view explode(friends) f1 as fd) f2 where f1.name = f2.name and f1.fd < f2.fd))  ww group by ww.fs having sum(ww.num) = 0;

7、数据倾斜

7.1、定义

数据倾斜，即单个节点任务所处理的数据量远大于同类型任务所处理的数据量，导致该节点成为整个作业的瓶颈，这是分布式系统不可能避免的问题。
从本质来说，导致数据倾斜有两种原因：
- 一是任务读取大文件，最常见的就是读取压缩的不可分割的大文件。
- 二是任务需要处理大量相同键的数据。
  - 数据含有大量无意义的数据，例如空值(NULL)、空字符串等
  - 含有倾斜数据在进行聚合计算时无法聚合中间结果，大量数据都需要经过 Shuffle阶段的处理，引起数据倾斜。
  - 数据在计算时做多维数据集合，导致维度膨胀引起的数据倾斜。
  - 两表进行Join，都含有大量相同的倾斜数据键。

7.2、原因

当集群的数据量增长到一定规模，有些数据需要归档或者转储，这时候往往会对数据进行压缩;
当对文件使用GZIP压缩等不支持文件分割操作的压缩方式，在日后有作业涉及读取压缩后的文件时，该压缩文件只会被一个任务所读取。
如果该压缩文件很大，则处理该文件的Map需要花费的时间会远多于读取普通文件的Map时间，该Map任务会成为作业运行的瓶颈。

为免因不可拆分大文件而引发数据读取的倾斜，在数据压缩的时候可以采用bzip2和Zip支持文件分割的压缩算法,或者使用像orc、SequenceFile等列式存储

业务无关的数据引发的数据倾斜

实际业务中有些大量的NUL值或者一些无意义的数据参与到计算作业中，这些数据可能来自业务为上报或因数据规范将某类数据进行归一化变成空值或空字符等形式，这些与业务无关的数据导致在进行分组聚合或者在执行表连接时发生数据倾斜。对于这类问题引发的数据倾斜，在计算过程中排除含有这类“异常”据即可。

多维聚合计算数据膨胀引起的数据倾斜

在多维聚合计算时存在这样的场景: select a，b，c， count() I from T group by a，b， c with rollup。对于上述的SOL，可以拆解成4种类型的键进行分组聚合，它们分别是(a，b，nul)、(a，b、nul)(a，nul，nul)和(nul，nul，nul)

如果T表的数据量很大，并且Map端的聚合不能很好地起到数据压缩的情况下，会导致Map端产出的数据急速膨胀，这种情况容易导致作业内存溢出的异常。如果T表含有数据倾斜键，会加剧 Shuffle过程的数据倾斜。

对上述的情况我们会很自然地想到拆解上面的SQL语句，将 rollup拆解成如下多个普通类型分组聚合的组合。

无法削减中间结果的数据量引发的数据倾斜

在一些操作中无法削减中间结果，例如使用 collect list聚合函数，存在如下SQL

select s_age， collect_list(s_score) list_score from student_tb_txt group by s_age

两个Hive数据表连接时引发的数据倾斜

两表进行普通的 repartition join时，如果表连接的键存在倾斜，那么在 Shufe阶段必然会引起数据倾斜。

遇到这种情况，Hve的通常做法还是启用两个作业，第一个作业处理没有便斜的数据，第二个作业将倾斜的数据存到分布式缓存中，分发到各个M即任务所在节点在Mip价段完成join操作，即 Mapjoin，这避免了 Shuffle，从而避免了数据倾斜。

8、Hive文件存储格式

原始的TEXT文本文件为585GB
Hive早期的RCFILE压缩后为505GB
Impala中的PARQUET压缩后为221GB
Hive中的ORC压缩后仅为131GB，压缩比最高。

8.1、文件存储方式

Hive支持行式存储格式和列式存储格式

行式存储格式：相关数据保存在一起，数据读取的时候以行为单位读取的
- 优点：这种存储格式简单、方便写入数据 。
- 缺点：不支持压缩、并且不支持列裁剪，数据分析开销较大
- 文件格式：TextFile、SequenceFile
列式存储格式：将不同的列存放在不同的块中。
- 优点：支持列裁剪、减少数据查询范围，数据支持压缩，节省空间。
- 缺点：写入数据相对困难，并且查询整行数据时，开销相对较大。
- 文件格式：ORC、PARQUET、RCFILE

8.2、文件存储格式

TextFile：

Hive默认格式，数据不做压缩，磁盘开销大，数据解析开销大。
可结合Gzip、Bzip2、Snappy等使用（系统自动检查，执行查询时自动解压）
但使用这种方式，hive不会对数据进行切分，从而无法对数据进行并行操作。
RCFile
- RCFile是Hive推出的一种专门面向列的数据格式。
- 它遵循“先按列划分，再垂直划分”的设计理念。
- 当查询过程中，针对它并不关心的列时，它会在IO上跳过这些列。
- 需要说明的是，
  - RCFile在map阶段从远端拷贝仍然是拷贝整个数据块，
  - 并且拷贝到本地目录后RCFile并不是真正直接跳过不需要的列，
  - 而是通过扫描每一个row group的头部定义来实现的，
  - 但是在整个HDFS Block 级别的头部并没有定义每个列从哪个row group起始到哪个row group结束。
  - 所以在读取所有列的情况下，RCFile的性能反而没有SequenceFile高。
ORC

ORC的全称是(Optimized Row Columnar)，ORC文件格式是一种Hadoop生态圈中的列式存储格式，它的产生早在2013年初，最初产生自Apache Hive，用于降低Hadoop数据存储空间和加速Hive查询速度。和Parquet类似，它并不是一个单纯的列式存储格式，仍然是首先根据行组分割整个表，在每一个行组内进行按列存储。
1）Postscript: 提供了解释文件其余部分的必要信息，包括文件的页脚和元数据部分的长度，文件的版本以及使用的常规压缩类型（例如，none，zlib，LZO，LZ4，ZSTD或snappy）

2）File Footer: 包含文件正文的布局，类型架构信息，行数以及每个列的统计信息。

3）Stripe条带数据块：文件正文分为条带。每个条带都是自包含的，只能使用自己的字节与文件的Footer和Postscript结合使用。每个条带仅包含整行，因此行不会跨越条带边界。Stripes有三个部分：条带中行的一组索引，数据本身和条带页脚。索引和数据部分都按列分割，因此只需要读取所需列的数据。

4）列统计：列统计信息的目标是，对于每个列，编写器记录计数并根据其他有用字段的类型进行记录。对于大多数原始类型，它记录最小值和最大值; 对于数字类型，它还存储总和。从Hive 1.1.0开始，列统计信息还将通过设置hasNull标志来记录行组中是否存在任何空值。ORC的谓词下推使用hasNull标志来更好地回答’IS NULL’查询。真实列数据块，其中又分为Index data（记录每列的索引信息），Raw Data（记录原始数据），Stripe Footer（记录每列的统计信息，min/max/sum等）。

Stripe在默认情况下64MB。文件中的条带彼此独立，形成分布式工作的自然单元。在每个条带中，列彼此分开，因此Reader只能读取所需的列。

8.3、文件压缩练习

TextFile

create table t_stored_text (
c1 string,
c2 string,
c3 string,
c4 string,
c5 string,
c6 string,
c7 string
)ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
STORED AS TEXTFILE ;load data local inpath '/yjx/test.data' into table t_stored_text ;dfs -du -h /hive/warehouse/yjx.db/t_stored_text;

ORC

create table t_stored_orc (
c1 string,
c2 string,
c3 string,
c4 string,
c5 string,
c6 string,
c7 string
)ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
STORED AS ORC ;load data local inpath '/yjx/test.data' into table t_stored_orc ;dfs -du -h /hive/warehouse/yjx.db/t_stored_orc;

9、Hive参数

9.1、设置参数的三种方式

hive当中的参数、变量，都是以命名空间开头

命名空间	读写权限	含义
hiveconf	可读写	hive_site当中的各配置变量
system	可读写	系统变量，包含JVM运行参数等，例如：system:user.name=root
env	只读	环境变量，例如：env:JAVA_HOME
hivevar	可读写	sql中直接使用的变量，例如：hive -d val=key

通过**${}方式进行引用**，其中system、env下的变量必须以前缀开头

配置文件方式

默认配置文件：hive-default.xml
用户自定义配置文件：${HIVE_HOME}/conf/hive-site.xml
注意：用户自定义配置会覆盖默认配置。另外，Hive也会读入Hadoop的配置，因为Hive是作为Hadoop的客户端启动的，Hive的配置会覆盖Hadoop的配置。配置文件的设定对本机启动的所有Hive进程都有效。
命令行参数方式
启动Hive时，可以在命令行添加-hiveconf param=value来设定参数。
例如：

beeline -u jdbc:hive2://yjx103:10000 -n root -hiveconf mapred.reduce.tasks=10;

注意：仅对本次hive启动有效
查看参数设置

set mapred.reduce.tasks;显示为：mapred.reduce.tasks=10默认mapred.reduce.tasks=-1

参数声明方式
可以在HQL中使用SET关键字设定参数
例如

set mapred.reduce.tasks=3;

注意：仅对本次hive启动有效。
查看参数设置

set mapred.reduce.tasks;

上述三种设定方式的优先级依次递增。即配置文件<命令行参数<参数声明。注意某些系统级的参数，例如log4j相关的设定，必须用前两种方式设定，因为那些参数的读取在会话建立以前已经完成了。

hive参数初始化配置

当前用户家目录下的.hiverc文件。如: ~/.hiverc

如果没有，可直接创建该文件，将需要设置的参数写到该文件中，hive启动运行时，会加载改文件中的配置。

9.2、常用的hive设置

0-hive.fetch.task.conversion=more;将hive拉取的模式设置为more模式
1-hive.exec.mode.local.auto 决定 Hive 是否应该自动地根据输入文件大小，在本地运行（在GateWay运行） ;
2-hive.auto.convert.join ：是否根据输入小表的大小，自动将 Reduce 端的 Common Join 转化为 Map Join，从而加快大表关联小表的 Join 速度。 默认：false。
3-mapred.reduce.tasks ：所提交 Job 的 reduer 的个数，使用 Hadoop Client 的配置。 默认是-1，表示Job执行的个数交由Hive来分配；
mapred.map.tasks:设置提交Job的map端个数；
4-hive.map.aggr=true 开启map端聚合；
hive.groupby.skewindata=true ：决定 group by 操作是否支持倾斜的数据。
原理是，在Group by中，对一些比较小的分区进行合并，默认是false；
5-hive.merge.mapredfiles ：是否开启合并 Map/Reduce 小文件，对于 Hadoop 0.20 以前的版本，起一首新的 Map/Reduce Job，对于 0.20 以后的版本，则是起使用 CombineInputFormat 的 MapOnly Job。 默认是：false；
6-hive.mapred.mode ：Map/Redure 模式，如果设置为 strict，将不允许笛卡尔积。 默认是：'nonstrict'；
7-hive.exec.parallel ：是否开启 map/reduce job的并发提交。
默认Map/Reduce job是顺序执行的，默认并发数量是8，可以配置。默认是：false；
8-hive.exec.dynamic.partition =true：是否打开动态分区。 需要打开，默认：false；
set hive.exec.dynamic.partition.mode=nonstirct

10、Hive企业优化

Hive优化的核心思想：把Hive SQL 当做Mapreduce程序去优化。
根据不通案例来不同优化

10.1、Fetch

我们之前是不是说过select * from 表；这样的sql语句是不会产生MR任务的，这涉及到一个fetch（抓取）的概念。
hive-default.xml.template hive.fetch.task.conversion–>
- 默认more 如果 none那么每次执行select * from 都是执行MapReduce

10.2、本地模式

hive.exec.mode.local.auto=true

10.3、并行执行

set hive.exec.parallel=true; //打开任务并行执行

set hive.exec.parallel.thread.number=16; //同一个sql允许最大并行度

10.4、严格模式

开启严格模式需要修改hive.mapred.mode值为strict，开启严格模式可以禁止3种类型的查询。

<property><name>hive.mapred.mode</name><value>strict</value>
</property>

防止用户执行低效率的SQL查询
- 对于分区表，除非where语句中含有分区字段过滤条件来限制范围，否则不允许执行
- 对于使用了order by语句的查询，要求必须使用limit语句
- 限制笛卡尔积的查询。

10.5、JVM重用

JVM重用可以使得JVM实例在同一个job中重新使用N次。
N的值可以在Hadoop的mapred-site.xml文件中进行配置。
通常在10-20之间，具体多少需要根据具体业务场景测试得出。

<property><name>mapreduce.job.jvm.numtasks</name><value>10</value>
</property>

很难避免小文件的场景或task特别多的场景，这类场景大多数执行时间都很短

10.6、表的优化（小表与大表）

Hive默认第一个(左面的)表是小表，然后将其存放到内存中，然后去与第二张表进行比较
现在优化后小表前后无所谓

10.6、表的优化（大表与大表）

针对于空值，可以将空值随机设置一个不影响结果的值
将来reduce的时候可以分区到不同的reduce，减少压力

10.7、mapside聚合

默认情况下，Map阶段同一Key数据分发给一个reduce，当一个key数据过大时就倾斜了。
并不是所有的聚合操作都需要在Reduce端完成，很多聚合操作都可以先在Map端进行部分聚合，最后在Reduce端得出最终结果。
- 开启Map端聚合参数设置
  - 是否在Map端进行聚合，默认为True
  - hive.map.aggr = true
- 在Map端进行聚合操作的条目数目
  - hive.groupby.mapaggr.checkinterval = 100000
- 有数据倾斜的时候进行负载均衡（默认是false）
  - hive.groupby.skewindata = true
  - 当选项设定为 true，生成的查询计划会有两个MR Job。
  - 分两次进行mapredue,第一次随机分获取中间结果,第二次正常分，获取最终结果

10.8、Count(Distinct)

防止所有的数据都分到一个Reduce上面
首先使用Group By对数据进行分组，然后在统计

10.9、防止笛卡尔积

行列过滤(列裁剪)

当表关联的时候，优先使用子查询对表的数据进行过滤,这样前面表关联数据就是少的，减少关联的次数

案例实操：

1．测试先关联两张表，再用where条件过滤

select o.id from bigtable b join ori o on o.id = b.id where o.id <= 10;

2．通过子查询后，再关联表

select b.id from bigtable b join (select id from ori where id <= 10 ) o on b.id = o.id;

工作上经常用到的hivesql小技巧
- 多表关联时要多用union all ,多表union all会优化成一个job。
- 修改表结构要用REPLACE COLUMNS。使用add colmns数据为空，这时因为元数据库没有更新
- 注意Hive中NULL和’ ‘的区别，不管哪一种做相加的时候都会变成null
- 去列参数set hive.support.quoted.identifiers=none; (membership_level|extra_info)?+.+
- 查询切记加limit