Hive函数、压缩存储、调优汇总、权限管理

左手的ㄟ右手 2024-04-17 16:50 12阅读 0赞

# Hive函数、压缩存储、调优、权限 #

### 一、函数 ###

### 1 系统内置函数 ###

显示自带函数的用法 desc function 函数名;

详细显示自带函数的用法 desc function extended 函数名;

### 2 自定义函数 ###

1）Hive 自带了一些函数，比如：max/min等，但是数量有限，自己可以通过自定义UDF来方便的扩展

2）当Hive提供的内置函数无法满足你的业务处理需要时，此时就可以考虑使用用户自定义函数

3）根据用户自定义函数类别分为以下三种：  
 UDF（User-Defined-Function） 一进一出  
 UDAF（User-Defined Aggregation Function） 多进一出，聚集函数 count/max/min  
 UDTF（User-Defined Table-Generating Functions） 一进多出 lateral view explore()  
  
4）官方文档地址  
 https://cwiki.apache.org/confluence/display/Hive/HivePlugins

5）编程步骤：  
 （1）继承org.apache.hadoop.hive.ql.UDF  
 （2）需要实现evaluate函数；evaluate函数支持重载；  
 （3）在hive的命令行窗口创建函数  
 a）添加jar  
 add jar linux\_jar\_path  
 b）创建function，  
 create \[temporary\] function \[dbname.\]function\_name AS class\_name;  
 （4）在hive的命令行窗口删除函数  
 Drop \[temporary\] function \[if exists\]\[dbname.\]function\_name;

6）注意事项  
 UDF必须要有返回类型，可以返回null，但是返回类型不能为void；

### 3 自定义UDF函数 ###

1．创建一个Maven工程Hive

2．导入依赖

<dependencies>
    		
    		<dependency>
    			<groupId>org.apache.hive</groupId>
    			<artifactId>hive-exec</artifactId>
    			<version>1.2.1</version>
    		</dependency>
    </dependencies>

3．创建一个类

package com.atguigu.hive;
    import org.apache.hadoop.hive.ql.exec.UDF;
    
    public class Lower extends UDF {
    
    	public String evaluate (final String s) {
    		
    		if (s == null) {
    			return null;
    		}
    		
    		return s.toLowerCase();
    	}
    }

4．打成jar包上传到服务器/opt/module/jars/udf.jar

5．将jar包添加到hive的classpath

add jar /opt/module/datas/udf.jar;

6．创建临时函数与开发好的java class关联

create temporary function mylower as “com.atguigu.hive.Lower”;

7．即可在hql中使用自定义的函数strip

select ename, mylower(ename) lowername from emp;

## 二、压缩和存储 ##

### 1 压缩 ###

压缩相关配置在doc文件中

<table> 
 <thead> 
  <tr> 
   <th>压缩算法</th> 
   <th>原始文件大小</th> 
   <th>压缩文件大小</th> 
   <th>压缩速度</th> 
   <th>解压速度</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>gzip</td> 
   <td>8.3GB</td> 
   <td>1.8GB</td> 
   <td>17.5MB/s</td> 
   <td>58MB/s</td> 
  </tr> 
  <tr> 
   <td>bzip2</td> 
   <td>8.3GB</td> 
   <td>1.1GB</td> 
   <td>2.4MB/s</td> 
   <td>9.5MB/s</td> 
  </tr> 
  <tr> 
   <td>LZO</td> 
   <td>8.3GB</td> 
   <td>2.9GB</td> 
   <td>49.3MB/s</td> 
   <td>74.6MB/s</td> 
  </tr> 
 </tbody> 
</table>

### 2 文件存储格式 ###

Hive支持的存储数的格式主要有：TEXTFILE 、SEQUENCEFILE、ORC、PARQUET

#### 1）列式存储和行式存储 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70]

**行存储的特点**

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

TEXTFILE 和 SEQUENCEFILE 的存储格式都是基于行存储的

**列存储的特点**

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

ORC 和 PARQUET 是基于列式存储的

**Tips：**

ORC并非完全基于列存储，因为其大小是250M一块(stripe)，在该stripe内是基于一定的规则存储

#### 2）TextFile格式 ####

默认格式，数据不做压缩，磁盘开销大，数据解析开销大

可结合Gzip、Bzip2使用，但使用Gzip这种方式，hive不会对数据进行切分，从而无法对数据进行并行操作

#### 3）Orc格式 ####

\[外链图片转存失败(img-QXf1Lhik-1566997532252)(C:\\Users\\86158\\AppData\\Roaming\\Typora\\typora-user-images\\1566996805074.png)\]

每个Orc文件由1个或多个stripe组成，每个stripe250MB大小，这个Stripe实际相当于RowGroup概念，不过大小由4MB->250MB，这样应该能提升顺序读的吞吐率

每个Stripe里有三部分组成，分别是Index Data，Row Data，Stripe Footer

*  Index Data：一个轻量级的index，默认是每隔1W行做一个索引。这里做的索引应该只是记录某行的各字段在Row Data中的offset
 *  Row Data：存的是具体的数据，先取部分行，然后对这些行按列进行存储。对每个列进行了编码，分成多个Stream来存储
 *  Stripe Footer：存的是各个Stream的类型，长度等信息

#### 4）Parquet格式 ####

Parquet文件是以二进制方式存储的，所以是不可以直接读取的，文件中包括该文件的数据和元数据，因此Parquet格式文件是自解析的

通常情况下，在存储Parquet数据的时候会按照Block大小设置行组的大小，由于一般情况下每一个Mapper任务处理数据的最小单位是一个Block，这样可以把每一个行组由一个Mapper任务处理，增大任务执行并行度

在Parquet中，有三种类型的页：数据页、字典页和索引页。数据页用于存储当前行组中该列的值，字典页存储该列值的编码字典，每一个列块中最多包含一个字典页，索引页用来存储当前行组下该列的索引，目前Parquet中还不支持索引页。

### 3 主流文件存储格式对比 ###

经过大量数据的验证发现：

存储文件压缩比率：ORC>Parquet>TextFile

存储文件查询速度相近

## 三、优化 ##

分区分桶也是hive优化的一种

**核心思想：把Hive SQL 当做Mapreduce程序去优化**

### 1 Explain 显示执行计划 ###

执行语句：

EXPLAIN \[EXTENDED\] query

eg：explain select \* from tablename

解析展示某表经过MR的流程，选择较为简单的流程开发方式，效率更高

### 2 开启本地模式 ###

本地模式数据不经过HDFS，效率更高，适用于 数据量比较小、测试数据

通过设置以下参数开启本地模式：

set hive.exec.mode.local.auto=true;

注意：

hive.exec.mode.local.auto.inputbytes.max默认值为128M

表示加载文件的最大值，若大于该配置仍会以集群方式来运行

\[外链图片转存失败(img-yfAqF6wz-1566997532252)(C:\\Users\\86158\\AppData\\Roaming\\Typora\\typora-user-images\\1561720449456.png)\]

### 3 表的优化 ###

#### 1）小表join大表 大表join小表 ####

旧版本小表在左，可以减少内存溢出错误发生几率，小表加载内存可以在Map端完成reduce

新版本的Hive底层已经对表的join做过优化，所以谁在左右无关

#### 2）大表join大表 ####

空key问题

解决方案：将空key直接过滤、空key转换赋随机值

*  过滤空key后，再执行join操作执行效率高
 *  操作
    
    随机分布空null值
    
    （1）设置5个reduce个数
    
    set mapreduce.job.reduces = 5;
    
    （2）JOIN两张表
    
        容易产生数据倾斜
        insert overwrite table jointable
        select n.* from nullidtable n left join ori b on n.id = b.id;
    
        不易发生数据倾斜concat
        insert overwrite table table1
        select n.* from nullidtable n full join ori o on 
        case when n.id is null then concat('hive', rand()) else n.id end = o.id;

#### 3）MapJoin ####

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

在Map 端完成join

1 SQL方式，在SQL语句中添加MapJoin标记(mapjoin hint)

2 开启自动的MapJoin

通过设置以下参数开启自动的MapJoin：

set hive.auto.convert.join = true;

大表小表判断的依据：

hive.mapjoin.smalltable.filesize=25000000 即25M

#### 4）group by ####

默认情况下，Map阶段同一Key数据分发给一个reduce，当一个key数据过大时就倾斜了。

并不是所有的聚合操作都需要在Reduce端完成，很多聚合操作都可以先在Map端进行部分聚合，最后在Reduce端得出最终结果。

开启Map端聚合参数设置

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

hive.map.aggr = true

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

hive.groupby.mapaggr.checkinterval = 100000

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

hive.groupby.skewindata = true

当选项设定为 true，生成的查询计划会有两个MR Job,是相同的Group By Key有可能被分发到不同的Reduce

#### 5） Count(Distinct) 去重统计 ####

数据量小的时候无所谓，数据量大的情况下，由于COUNT DISTINCT操作需要用一个Reduce Task来完成，这一个Reduce需要处理的数据量太大，就会导致整个Job很难完成，容易产生数据倾斜

一般COUNT DISTINCT使用先GROUP BY再COUNT的方式替换

查询的2种方法

记得设置reduce数

采用去重ID查询： select count(distinct id) from bigtable;

采用group by去重ID查询：select count(id) from (select id from bigtable group by id) a;

#### 6）笛卡尔积 ####

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

#### 7）行列过滤 ####

谓词下推，先过滤

列处理：在SELECT中，只拿需要的列，如果有，尽量使用分区过滤，少用SELECT \*

行处理：在分区剪裁中，当使用外关联时，如果将副表的过滤条件写在Where后面，那么就会先全表关联，之后再过滤

#### 8）动态分区 ####

关系型数据库中，对分区表Insert数据时候，数据库自动会根据分区字段的值，将数据插入到相应的分区中，

Hive中也提供了类似的机制，即动态分区(Dynamic Partition)，只不过，使用Hive的动态分区，需要进行相应的配置

##### 1 开启动态分区参数设置 #####

（1）开启动态分区功能（默认true，开启）  
 hive.exec.dynamic.partition=true  
（2）设置为非严格模式  
动态分区的模式，默认strict，表示必须指定至少一个分区为静态分区，nonstrict模式表示允许所有的分区字段都可以使用动态分区  
 hive.exec.dynamic.partition.mode=nonstrict  
（3）在所有执行MR的节点上，最大一共可以创建多少个动态分区  
 hive.exec.max.dynamic.partitions=1000  
（4）在每个执行MR的节点上，最大可以创建多少个动态分区。该参数需要根据实际的数据来设定。  
比如：源数据中包含了一年的数据，即day字段有365个值，那么该参数就需要设置成大于365，如果使用默认值100，则会报错  
 hive.exec.max.dynamic.partitions.pernode=100  
（5）整个MR Job中，最大可以创建多少个HDFS文件  
 hive.exec.max.created.files=100000  
（6）当有空分区生成时，是否抛出异常。一般不需要设置  
 hive.error.on.empty.partition=false

##### 2 创建动态分区步骤 #####

**1) 创建分区表**

CREATE TABLE psn01(  
id int,  
name string,  
likes ARRAY < string >,  
address MAP < string,string >  
)  
PARTITIONED BY(sex string,age int)  
ROW FORMAT DELIMITED  
FIELDS TERMINATED BY ‘,’  
COLLECTION ITEMS TERMINATED BY ‘-’  
MAP KEYS TERMINATED BY ‘:’;

**2) 创建普通表**

CREATE TABLE psn02(  
id int,  
name string,  
sex string,  
age int,  
likes ARRAY < string >,  
address MAP < string,string >  
)  
ROW FORMAT DELIMITED  
FIELDS TERMINATED BY ‘,’  
COLLECTION ITEMS TERMINATED BY ‘-’  
MAP KEYS TERMINATED BY ‘:’;

**3) 将数据传到普通表**

load data local inpath ‘/root/data2’ into table psn02;

**4) 将普通表数据加载到分区表**

from psn02  
insert overwrite table psn01 partition(sex,age)  
select id, name, likes, address , sex, age distribute by sex,age;

#### 9）分区/分桶 ####

数据量过大的时候是提高效率

### 4 数据倾斜 ###

#### 1 ) 合理设置Map数 ####

1）通常情况下，作业会通过input的目录产生一个或者多个map任务

主要的决定因素有：input的文件总个数，input的文件大小，集群设置的文件块大小。

2）是不是map数越多越好？

答案是否定的。如果一个任务有很多小文件（远远小于块大小128m），则每个小文件也会被当做一个块，用一个map任务来完成，而一个map任务启动和初始化的时间远远大于逻辑处理的时间，就会造成很大的资源浪费。而且，同时可执行的map数是受限的。

3）是不是保证每个map处理接近128m的文件块，就高枕无忧了？

答案也是不一定。比如有一个127m的文件，正常会用一个map去完成，但这个文件只有一个或者两个小字段，却有几千万的记录，如果map处理的逻辑比较复杂，用一个map任务去做，肯定也比较耗时。

针对上面的问题2和3，我们需要采取两种方式来解决：即减少map数和增加map数；

#### 2) 小文件进行合并 ####

在map执行前合并小文件，减少map数：CombineHiveInputFormat具有对小文件进行合并的功能（系统默认的格式）

HiveInputFormat没有对小文件合并功能。

set hive.input.format= org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

#### 3) 复杂文件增加Map数 ####

当input的文件都很大，任务逻辑复杂，map执行非常慢的时候，可以考虑增加Map数，来使得每个map处理的数据量减少，从而提高任务的执行效率。

**Tips:**

**map、split、reduce在 hive中默认大小都是250M**

增加map的方法为：

computeSliteSize(Math.max(minSize,Math.min(maxSize,blocksize)))=blocksize=128M公式，调整maxSize最大值。让maxSize最大值低于blocksize就可以增加map的个数。

设置最大切片值为100个字节

set mapreduce.input.fileinputformat.split.maxsize=100

#### 4) 合理设置Reduce数 ####

1．调整reduce个数方法一

（1）每个Reduce处理的数据量默认是256MB

hive.exec.reducers.bytes.per.reducer=256000000

（2）每个任务最大的reduce数，默认为1009

hive.exec.reducers.max=1009

（3）计算reducer数的公式

N=min(参数2，总输入数据量/参数1)

2．调整reduce个数方法二

在hadoop的mapred-default.xml文件中修改

设置每个job的Reduce个数

set mapreduce.job.reduces = 15;

3．reduce个数并不是越多越好

1）过多的启动和初始化reduce也会消耗时间和资源；

2）另外，有多少个reduce，就会有多少个输出文件，如果生成了很多个小文件，那么如果这些小文件作为下一个任务的输入，则也会出现小文件过多的问题；

在设置reduce个数的时候也需要考虑这两个原则：

处理大数据量利用合适的reduce数

使单个reduce任务处理数据量大小要合适

#### 5) 控制Map Reduce数量 ####

**Map数量相关的参数**（一般不做修改）

一个split的最大值，即每个map处理文件的最大值

mapred.max.split.size

一个节点上split的最小值

mapred.min.split.size.per.node

一个机架上split的最小值

mapred.min.split.size.per.rack

**Reduce数量相关的参数**

强制指定reduce任务的数量（默认1）

mapred.reduce.tasks

每个reduce任务处理的数据量

hive.exec.reducers.bytes.per.reducer

每个任务最大的reduce数

hive.exec.reducers.max

### 5 并行执行 ###

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

特点：

同时运行多个job任务，速度快但是占用内存资源

通过设置以下参数开启并行模式

set hive.exec.parallel=true;

一次SQL计算中允许并行执行的job个数的最大值，默认开启并行后，并行度是8

hive.exec.parallel.thread.number=8

### 6 严格模式 ###

Hive提供了一个严格模式，可以防止用户执行那些可能意想不到的不好的影响的查询，防止大面积执行语句，防止有误操作

通过设置以下参数开启严格模式：

set hive.mapred.mode=strict;

（默认为：nonstrict非严格模式）

**查询限制**

1、对于分区表，必须添加where对于分区字段的条件过滤，否则不允许执行，----用户不允许扫描所有分区

省了一个分区的全扫描，只展示where字段分区的标准

2、order by 查询语句必须使用 limit 语句输出限制 —一个Reducer中进行数据汇总处理

order by 默认是在reduce端进行排序，换而言之，是在reduce端汇总数据

3、限制执行笛卡尔积的查询

### 5 HIVE排序 ###

**Order By** — 对于查询结果做全排序，只允许有一个reduce处理

（当数据量较大时，应慎用。严格模式下，必须结合limit来使用）

**Sort By** — 对于单个reduce的数据进行排序

**Distribute By** — 分区排序，经常和Sort By结合使用

**Cluster By** — 相当于 Sort By + Distribute By

（Cluster By不能通过asc、desc的方式指定排序规则；

可通过 distribute by column sort by column asc|desc 的方式）

### 6 MAP-Side聚合 ###

通过设置以下参数开启Map端的聚合：

set hive.map.aggr = true;

对group by 产生的数据倾斜做优化

hive.groupby.skewindata

（默认false）

在分组不合理的情况下，上面的语句执行原理是：在map阶段执行两个聚合

第一个map拿到数据后进行随机分发输出给每个reduce，此时reduce做局部聚合拿到count结果，第二次再按照普通的规则继续下部操作

### 8 HIVE-JVM重用 ###

JVM重用是Hadoop调优参数的内容，其对Hive的性能具有非常大的影响，特别是对于很难避免小文件的场景或task特别多的场景，这类场景大多数执行时间都很短。

Hadoop的默认配置通常是使用派生JVM来执行map和Reduce任务的。这时JVM的启动过程可能会造成相当大的开销，尤其是执行的job包含有成百上千task任务的情况。

JVM重用可以使得JVM实例在同一个job中重新使用N次。

N的值可以在Hadoop的mapred-site.xml文件中进行配置。通常在10-20之间

<property>
      <name>mapreduce.job.jvm.numtasks</name>
      <value>10</value>
      <description>How many tasks to run per jvm. If set to -1, there is no limit.</description>
    </property>

这个功能的缺点是：

设置开启之后，task插槽会一直占用资源，不论是否有task运行，直到所有的task即整个job全部执行完成时，才会释放所有的task插槽资源！

如果此时有数据倾斜，则会一致影响到整个job任务结束

直接申请多个jvm运行空间

适用场景：1、小文件个数过多 2、task个数过多

通过 set mapred.job.reuse.jvm.num.tasks=n; 来设置

—n 为task插槽个数/JVM数

### 9 推测执行 ###

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

设置开启推测执行参数：Hadoop的mapred-site.xml文件中进行配置

<property>
      <name>mapreduce.map.speculative</name>
      <value>true</value>
      <description>If true, then multiple instances of some map tasks may be executed in parallel.</description>
    </property>
    
    <property>
      <name>mapreduce.reduce.speculative</name>
      <value>true</value>
      <description>If true, then multiple instances of some reduce tasks may be executed in parallel.</description>
    </property>

不过hive本身也提供了配置项来控制reduce-side的推测执行：

<property>
        <name>hive.mapred.reduce.tasks.speculative.execution</name>
        <value>true</value>
        <description>Whether speculative execution for reducers should be turned on. </description>
      </property>

注意：

如果用户因为输入数据量很大而需要执行长时间的map或者Reduce task的话，那么启动推测执行造成的浪费是非常巨大大。

### 10 压缩 ###