Hive最全总结

Hive 架构与原理

Hive架构

1 用户接口&＃xff1a;Client

CLI&＃xff08;command-line interface&＃xff09;、JDBC/ODBC(jdbc访问hive)

2 元数据&＃xff1a;Metastore

元数据包括&＃xff1a;表名、表所属的数据库&＃xff08;默认是default&＃xff09;、表的拥有者、列/分区字段、表的类型&＃xff08;是否是外部表&＃xff09;、表的数据所在目录等&＃xff1b;默认存储在自带的derby数据库中&＃xff0c;推荐使用MySQL存储Metastore

3 Hadoop

使用HDFS进行存储&＃xff0c;使用MapReduce进行计算。

4 驱动器&＃xff1a;Driver

&＃xff08;1&＃xff09;解析器&＃xff08;SQL Parser&＃xff09;&＃xff1a;将SQL字符串转换成抽象语法树AST&＃xff0c;这一步一般都用第三方工具库完成&＃xff0c;比如antlr&＃xff1b;对AST进行语法分析&＃xff0c;比如表是否存在、字段是否存在、SQL语义是否有误。

&＃xff08;2&＃xff09;编译器&＃xff08;Physical Plan&＃xff09;&＃xff1a;将AST编译生成逻辑执行计划。

&＃xff08;3&＃xff09;优化器&＃xff08;Query Optimizer&＃xff09;&＃xff1a;对逻辑执行计划进行优化。

&＃xff08;4&＃xff09;执行器&＃xff08;Execution&＃xff09;&＃xff1a;把逻辑执行计划转换成可以运行的物理计划。对于Hive来说&＃xff0c;就是MR/Spark。

Hive运行原理

Hive通过给用户提供的一系列交互接口&＃xff0c;接收到用户的指令(SQL)&＃xff0c;使用自己的Driver&＃xff0c;结合元数据(MetaStore)&＃xff0c;将这些指令翻译成MapReduce&＃xff0c;提交到Hadoop中执行&＃xff0c;最后&＃xff0c;将执行返回的结果输出到用户交互接口。

其实&＃xff0c;还可以这样理解&＃xff1a;Hive要做的就是将SQL翻译成MapReduce程序代码。实际上&＃xff0c;Hive内置了很多Operator&＃xff0c;每个Operator完成一个特定的计算过程&＃xff0c;Hive将这些Operator构造成一个有向无环图DAG&＃xff0c;然后根据这些Operator之间是否存在shuffle将其封装到map或者reduce函数中&＃xff0c;之后就可以提交给MapReduce执行了。

内部表与外部表

不同点

&＃xff08;1&＃xff09; 外部表不会加载数据到Hive&＃xff0c;减少数据传输、数据还能共享。

共享的理解就是&＃xff1a;当我们删除一个内部表时&＃xff0c;Hive 也会删除这个表中数据。内部表不适合和其他工具共享数据。

&＃xff08;2&＃xff09;Hive创建内部表时&＃xff0c;会将数据移动到数据仓库指向的路径。

创建外部表时&＃xff0c;仅记录数据所在的路径&＃xff0c;不对数据的位置做任何改变&＃xff0c;

在删除表的时候&＃xff0c;内部表的元数据和数据会被一起删除&＃xff0c;而外部表只删除元数据&＃xff0c;不删除数据。这样外部表相对来说更加安全些&＃xff0c;数据组织也更加灵活&＃xff0c;方便共享源数据。

场景选择

在公司中绝大多数场景都是外部表。

自己使用的临时表&＃xff0c;才会创建内部表。

Hive分区与分桶

六千字总结:大数据框架(分区&＃xff0c;分桶&＃xff0c;分片),建议收藏

Hive分区

是按照数据表的某列或者某些列分为多区&＃xff0c;在hive存储上是hdfs文件&＃xff0c;也就是文件夹形式。现在最常用的跑T&＃43;1数据&＃xff0c;按当天时间分区的较多。

把每天通过sqoop或者datax拉取的一天的数据存储一个区&＃xff0c;也就是所谓的文件夹与文件。在查询时只要指定分区字段的值就可以直接从该分区查找即可。创建分区表的时候&＃xff0c;要通过关键字 partitioned by &＃xff08;column name  string&＃xff09;声明该表是分区表&＃xff0c;并且是按照字段column name进行分区&＃xff0c;column name值一致的所有记录存放在一个分区中&＃xff0c;分区属性name的类型是string类型。

当然&＃xff0c;可以依据多个列进行分区&＃xff0c;即对某个分区的数据按照某些列继续分区。

向分区表导入数据的时候&＃xff0c;要通过关键字partition&＃xff08;&＃xff08;column name&＃61;"xxxx"&＃xff09;显示声明数据要导入到表的哪个分区

设置分区的影响

1. 首先是hive本身对分区数有限制&＃xff0c;不过可以修改限制的数量。

set hive.exec.dynamic.partition&＃61;true;
set hive.exec.max.dynamic.partitions&＃61;1000; 
set hive.exec.dynamic.partition.mode&＃61;nonstrict; 
set hive.exec.parallel.thread.number&＃61;264;

1. hdfs对单个目录下的目录数量或者文件数量也是有限制的&＃xff0c;也是可以修改的。
2. NN的内存肯定会限制&＃xff0c;这是最重要的&＃xff0c;如果分区数很大&＃xff0c;会影响NN服务&＃xff0c;进而影响一系列依赖于NN的服务。所以最好合理设置分区规则&＃xff0c;对小文件也可以定期合并&＃xff0c;减少NN的压力。

Hive的分桶

在分区数量过于庞大以至于可能导致文件系统崩溃时&＃xff0c;我们就需要使用分桶来解决问题

分桶是相对分区进行更细粒度的划分。分桶则是指定分桶表的某一列&＃xff0c;让该列数据按照哈希取模的方式随机、均匀的分发到各个桶文件中。因为分桶操作需要根据某一列具体数据来进行哈希取模操作&＃xff0c;故指定的分桶列必须基于表中的某一列&＃xff08;字段&＃xff09; 要使用关键字clustered by 指定分区依据的列名&＃xff0c;还要指定分为多少桶&＃xff1a;

create table test(id int,name string) cluster by (id) into 5 buckets .......

insert into buck select id ,name from p cluster by (id)

Hive分区分桶区别

1. 分区是表的部分列的集合&＃xff0c;可以为频繁使用的数据建立分区&＃xff0c;这样查找分区中的数据时就不需要扫描全表&＃xff0c;这对于提高查找效率很有帮助。
2. 不同于分区对列直接进行拆分&＃xff0c;桶往往使用列的哈希值对数据打散&＃xff0c;并分发到各个不同的桶中从而完成数据的分桶过程。
3. 分区和分桶最大的区别就是分桶随机分割数据库&＃xff0c;分区是非随机分割数据库。

函数

本环节不再介绍简单的函数&＃xff0c;比如&＃xff1a;&＃39;if&＃39; &＃xff0c;&＃39;is not null&＃39; ,&＃39;&＃61;&＃61;&＃39;等等这类的函数。

内置函数

&＃xff08;1&＃xff09; NVL

给值为NULL的数据赋值&＃xff0c;它的格式是NVL( value&＃xff0c;default_value)。它的功能是如果value为NULL&＃xff0c;则NVL函数返回default_value的值&＃xff0c;否则返回value的值&＃xff0c;如果两个参数都为NULL &＃xff0c;则返回NULL

select nvl(column, 0) from xxx&＃xff1b;

&＃xff08;2&＃xff09;行转列

&＃xff08;3&＃xff09;列转行(一列转多行)

Split(str, separator)&＃xff1a; 将字符串按照后面的分隔符切割&＃xff0c;转换成字符array。

EXPLODE(col)&＃xff1a;将hive一列中复杂的array或者map结构拆分成多行。

LATERAL VIEW

用法&＃xff1a;

LATERAL VIEW udtf(expression) tableAlias AS columnAlias

解释&＃xff1a;lateral view用于和split, explode等UDTF一起使用&＃xff0c;它能够将一行数据拆成多行数据&＃xff0c;在此基础上可以对拆分后的数据进行聚合。

lateral view首先为原始表的每行调用UDTF&＃xff0c;UDTF会把一行拆分成一或者多行&＃xff0c;lateral view再把结果组合&＃xff0c;产生一个支持别名表的虚拟表。

准备数据源测试

SQL

SELECT movie,category_name 
FROM movie_info 
lateral VIEW
explode(split(category,",")) movie_info_tmp  AS category_name ;

测试结果

《功勋》      记录
《功勋》      剧情
《战狼2》     战争
《战狼2》     动作
《战狼2》     灾难

窗口函数

&＃xff08;1&＃xff09;OVER()

定分析函数工作的数据窗口大小&＃xff0c;这个数据窗口大小可能会随着行的变而变化。

&＃xff08;2&＃xff09;CURRENT ROW&＃xff08;当前行&＃xff09;

语法

n PRECEDING&＃xff1a;往前n行数据

n FOLLOWING&＃xff1a;往后n行数据

&＃xff08;3&＃xff09;UNBOUNDED&＃xff08;无边界&＃xff09;

UNBOUNDED PRECEDING 前无边界&＃xff0c;表示从前面的起点

UNBOUNDED FOLLOWING后无边界&＃xff0c;表示到后面的终点

SQL案例&＃xff1a;由起点到当前行的聚合

select 
    sum(money) over(partition by user_id order by pay_time rows between UNBOUNDED PRECEDING and current row) 
from or_order;

SQL案例&＃xff1a;当前行和前面一行做聚合

select 
    sum(money) over(partition by user_id order by pay_time rows between 1 PRECEDING and current row) 
from or_order;

SQL案例&＃xff1a;当前行和前面一行和后一行做聚合

select 
    sum(money) over(partition by user_id order by pay_time rows between 1 PRECEDING AND 1 FOLLOWING )
from or_order;

SQL案例&＃xff1a;当前行及后面所有行

select 
    sum(money) over(partition by user_id order by pay_time rows between current row and UNBOUNDED FOLLOWING  )
from or_order;

&＃xff08;3&＃xff09;LAG(col,n,default_val)

往前第n行数据&＃xff0c;没有的话default_val

&＃xff08;4&＃xff09;LEAD(col,n, default_val)

往后第n行数据&＃xff0c;没有的话default_val

SQL案例&＃xff1a;查询用户购买明细以及上次的购买时间和下次购买时间

select 
 user_id,,pay_time,money,
 
 lag(pay_time,1,&＃39;1970-01-01&＃39;) over(PARTITION by name order by pay_time) prev_time,
 
 lead(pay_time,1,&＃39;1970-01-01&＃39;) over(PARTITION by name order by pay_time) next_time
from or_order;

&＃xff08;5&＃xff09;FIRST_VALUE(col,true/false)

当前窗口下的第一个值&＃xff0c;第二个参数为true&＃xff0c;跳过空值。

&＃xff08;6&＃xff09;LAST_VALUE (col,true/false)

当前窗口下的最后一个值&＃xff0c;第二个参数为true&＃xff0c;跳过空值。

SQL案例&＃xff1a;查询顾用户每个月第一次的购买时间 和 每个月的最后一次购买时间。

select
 FIRST_VALUE(pay_time) 
     over(
         partition by user_id,month(pay_time) order by pay_time 
         rows between UNBOUNDED PRECEDING and UNBOUNDED FOLLOWING
         ) first_time,
 
 LAST_VALUE(pay_time) 
     over(partition by user_id,month(pay_time) order by pay_time rows between UNBOUNDED PRECEDING and UNBOUNDED FOLLOWING
     ) last_time
from or_order;

&＃xff08;7&＃xff09;NTILE(n)

把有序窗口的行分发到指定数据的组中&＃xff0c;各个组有编号&＃xff0c;编号从1开始&＃xff0c;对于每一行&＃xff0c;NTILE返回此行所属的组的编号。&＃xff08;用于将分组数据按照顺序切分成n片&＃xff0c;返回当前切片值&＃xff09;

SQL案例&＃xff1a;查询前25%时间的订单信息

select * from (
    select User_id,pay_time,money,
    
    ntile(4) over(order by pay_time) sorted
    
    from or_order
) t
where sorted &＃61; 1;

4个By

&＃xff08;1&＃xff09;Order By

全局排序&＃xff0c;只有一个Reducer。

&＃xff08;2&＃xff09;Sort By

分区内有序。

&＃xff08;3&＃xff09;Distrbute By

类似MR中Partition&＃xff0c;进行分区&＃xff0c;结合sort by使用。

&＃xff08;4&＃xff09; Cluster By

当Distribute by和Sorts by字段相同时&＃xff0c;可以使用Cluster by方式。Cluster by除了具有Distribute by的功能外还兼具Sort by的功能。但是排序只能是升序排序&＃xff0c;不能指定排序规则为ASC或者DESC。

在生产环境中Order By用的比较少&＃xff0c;容易导致OOM。

在生产环境中Sort By&＃43; Distrbute By用的多。

排序函数

&＃xff08;1&＃xff09;RANK()

排序相同时会重复&＃xff0c;总数不会变

1
1
3
3
5

&＃xff08;2&＃xff09;DENSE_RANK()

排序相同时会重复&＃xff0c;总数会减少

1
1
2
2
3

&＃xff08;3&＃xff09;ROW_NUMBER()

会根据顺序计算

1
2
3
4
5

Hive 优化

首先要这样优化的原理&＃xff0c;再去适当去调节参数和选择方案。

1. 表的优化

&＃xff08;1&＃xff09; 小表、大表Join

将key相对分散&＃xff0c;并且数据量小的表放在join的左边&＃xff0c;这样可以有效减少内存溢出错误发生的几率&＃xff1b;再进一步&＃xff0c;可以使用map join让小的维度表&＃xff08;1000条以下的记录条数&＃xff09;先进内存。在map端完成reduce。

&＃xff08;2&＃xff09; 大表Join大表

a. 空key过滤

有时join超时是因为某些key对应的数据太多&＃xff0c;而相同key对应的数据都会发送到相同的reducer上&＃xff0c;从而导致内存不够。此时我们应该仔细分析这些异常的key&＃xff0c;很多情况下&＃xff0c;这些key对应的数据是异常数据&＃xff0c;我们需要在SQL语句中进行过滤。

b. 空key转换

有时虽然某个key为空对应的数据很多&＃xff0c;但是相应的数据不是异常数据&＃xff0c;必须要包含在join的结果中&＃xff0c;此时我们可以表a中key为空的字段赋一个随机的值&＃xff0c;使得数据随机均匀地分不到不同的reducer上。

&＃xff08;3&＃xff09; MapJoin

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

设置自动选择Mapjoin
set hive.auto.convert.join &＃61; true; 默认为true

大表小表的阈值设置&＃xff08;默认25M以下认为是小表&＃xff09;&＃xff1a;
set hive.mapjoin.smalltable.filesize&＃61;25000000;

&＃xff08;4&＃xff09; Group By

Map阶段同一Key数据分发给一个reduce&＃xff0c;当一个key数据过大时就倾斜了。并不是所有的聚合操作都需要在Reduce端完成&＃xff0c;很多聚合操作都可以先在Map端进行部分聚合&＃xff0c;最后在Reduce端得出最终结果。

&＃xff08;5&＃xff09; 开启Map端聚合

// 是否在Map端进行聚合&＃xff0c;默认为True
set hive.map.aggr &＃61; true

// 在Map端进行聚合操作的条目数目
set hive.groupby.mapaggr.checkinterval &＃61; 100000

// 有数据倾斜的时候进行负载均衡&＃xff08;默认是false&＃xff09;
set hive.groupby.skewindata &＃61; true

对数据倾斜负载均衡的理解

会有两个MR Job。第一个MR Job中&＃xff0c;Map的输出结果会随机分布到Reduce中&＃xff0c;每个Reduce做部分聚合操作&＃xff0c;并输出结果&＃xff0c;这样处理的结果是相同的Group By Key有可能被分发到不同的Reduce中&＃xff0c;从而达到负载均衡的目的&＃xff1b;第二个MR Job再根据预处理的数据结果按照Group By Key分布到Reduce中&＃xff08;这个过程可以保证相同的Group By Key被分布到同一个Reduce中&＃xff09;&＃xff0c;最后完成最终的聚合操作。

(6) Count(Distinct) 去重统计

由于COUNT DISTINCT操作需要用一个Reduce Task来完成&＃xff0c;这一个Reduce需要处理的数据量太大&＃xff0c;就会导致整个Job很难完成&＃xff0c;一般COUNT DISTINCT使用先GROUP BY再COUNT的方式替换,但是需要注意group by造成的数据倾斜问题。

(7) 笛卡尔积

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

(8) 行列过滤

「列处理」&＃xff1a;在SELECT中&＃xff0c;只拿需要的列&＃xff0c;如果有&＃xff0c;尽量使用分区过滤&＃xff0c;少用SELECT *。

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

2. 合理设置Map及Reduce数

首先理清楚Map数是越多越好吗&＃xff1f;

「逻辑」&＃xff1a;如果一个任务有很多小文件&＃xff08;远远小于块大小128m&＃xff09;&＃xff0c;则每个小文件也会被当做一个块&＃xff0c;用一个map任务来完成&＃xff0c;而一个map任务启动和初始化的时间远远大于逻辑处理的时间&＃xff0c;就会造成很大的资源浪费。

保证每个map处理接近128m的文件块是不是就可以了&＃xff1f;

「逻辑」&＃xff1a;比如有一个127m的文件&＃xff0c;正常会用一个map去完成&＃xff0c;但这个文件只有一个或者两个小字段&＃xff0c;却有几千万的记录&＃xff0c;如果map处理的逻辑比较复杂&＃xff0c;用一个map任务去做&＃xff0c;肯定也比较耗时

复杂文件增加Map数

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

computeSliteSize(Math.max(minSize,Math.min(maxSize,blocksize)))&＃61;blocksize&＃61;128M

调整maxSize最大值。让maxSize最大值低于blocksize就可以增加map的个数。

小文件进行合并&＃xff0c;减少map数

在map执行前合并小文件&＃xff0c;减少map数&＃xff1a;CombineHiveInputFormat具有对小文件进行合并的功能&＃xff08;系统默认的格式&＃xff09;。

set hive.input.format&＃61; org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

Map-Reduce的任务结束时合并小文件的设置

// 在map-only任务结束时合并小文件&＃xff0c;默认true
SET hive.merge.mapfiles &＃61; true;

// 在map-reduce任务结束时合并小文件&＃xff0c;默认false
SET hive.merge.mapredfiles &＃61; true;

// 合并文件的大小&＃xff0c;默认256M
SET hive.merge.size.per.task &＃61; 268435456;

//当输出文件的平均大小小于该值时&＃xff0c;启动一个独立的map-reduce任务进行文件merge
SET hive.merge.smallfiles.avgsize &＃61; 16777216;

3. 合理设置Reduce数

同样考虑是不是越多越好&＃xff1f;

过多的启动和初始化reduce也会消耗时间和资源。有多少个reduce&＃xff0c;就会有多少个输出文件&＃xff0c;如果生成了很多个小文件&＃xff0c;那么如果这些小文件作为下一个任务的输入&＃xff0c;则也会出现小文件过多的问题。

&＃xff08;1&＃xff09;数据量设置

// 每个Reduce处理的数据量默认是256MB
hive.exec.reducers.bytes.per.reducer&＃61;256000000

// 每个任务最大的reduce数&＃xff0c;默认为1009
hive.exec.reducers.max&＃61;1009

// 计算reducer数的公式
N&＃61;min(hive.exec.reducers.max&＃xff0c;总输入数据量/hive.exec.reducers.bytes.per.reducer)

&＃xff08;2&＃xff09;文件配置

mapreduce.job.reduces &＃61; 15

4. 并行执行

通过设置参数hive.exec.parallel值为true&＃xff0c;就可以开启并发执行。不过&＃xff0c;在共享集群中&＃xff0c;需要注意下&＃xff0c;如果job中并行阶段增多&＃xff0c;那么集群利用率就会增加。建议在数据量大,sql很长的时候使用,数据量小,sql比较的小开启有可能还不如之前快。

//打开任务并行执行&＃xff0c;默认为false
set hive.exec.parallel&＃61;true; 

//同一个sql允许最大并行度&＃xff0c;默认为8。
set hive.exec.parallel.thread.number&＃61;16;  

5. JVM重用

JVM来执行map和Reduce任务的。这时JVM的启动过程可能会造成相当大的开销&＃xff0c;尤其是执行的job包含有成百上千task任务的情况。JVM重用可以使得JVM实例在同一个job中重新使用N次。

缺点是&＃xff0c;开启JVM重用将一直占用使用到的task插槽&＃xff0c;以便进行重用&＃xff0c;直到任务完成后才能释放。

set mapreduce.job.jvm.numtasks&＃61;10

6. 列式存储

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

TEXTFILE和SEQUENCEFILE的存储格式都是基于行存储的&＃xff1b;

ORC和PARQUET是基于列式存储的。

7. 压缩&＃xff08;选择快的&＃xff09;

// 启用中间数据压缩
set hive.exec.compress.intermediate&＃61;true 

// 启用最终数据压缩
set mapreduce.map.output.compress&＃61;true 

// 设置压缩方式
set mapreduce.map.outout.compress.codec&＃61;

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

Hive数据倾斜

你的数据倾斜了吗&＃xff1f;一文帮你数据处理再均衡

Hive数据倾斜表现

就是单说hive自身的MR引擎&＃xff1a;发现所有的map task全部完成&＃xff0c;并且99%的reduce task完成&＃xff0c;只剩下一个或者少数几个reduce task一直在执行&＃xff0c;这种情况下一般都是发生了数据倾斜。说白了就是Hive的数据倾斜本质上是MapReduce的数据倾斜。

Hive数据倾斜的原因

在MapReduce编程模型中十分常见&＃xff0c;大量相同的key被分配到一个reduce里&＃xff0c;造成一个reduce任务累死&＃xff0c;其他reduce任务闲死。查看任务进度&＃xff0c;发现长时间停留在99%或100%&＃xff0c;查看任务监控界面&＃xff0c;只有少量的reduce子任务未完成。

1. key分布不均衡。
2. 业务问题或者业务数据本身的问题&＃xff0c;某些数据比较集中。

&＃xff08;1&＃xff09;join小表&＃xff1a;其中一个表是小表&＃xff0c;但是key比较集中&＃xff0c;导致的就是某些Reduce的值偏高。

&＃xff08;2&＃xff09;空值或无意义值&＃xff1a;如果缺失的项很多&＃xff0c;在做join时这些空值就会非常集中&＃xff0c;拖累进度。

&＃xff08;3&＃xff09;group by&＃xff1a;维度过小。

&＃xff08;4&＃xff09;distinct&＃xff1a;导致最终只有一个Reduce任务。

Hive数据倾斜解决

1. group by代替distinct 要统计某一列的去重数时&＃xff0c;如果数据量很大&＃xff0c;count(distinct)就会非常慢&＃xff0c;原因与order by类似&＃xff0c;count(distinct)逻辑导致最终只有一个Reduce任务。
2. 对1再优化&＃xff1a;group by配置调整

&＃xff08;1&＃xff09;map端预聚合

&＃xff08;2&＃xff09;group by时&＃xff0c;combiner在map端做部分预聚合&＃xff0c;可以有效减少shuffle数据量。

&＃xff08;3&＃xff09;checkinterval&＃xff1a;设置map端预聚合的行数阈值&＃xff0c;超过该值就会分拆job。

hive.map.aggr&＃61;true //默认

hive.groupby.mapaggr.checkinterval&＃61;100000 // 默认

&＃xff08;4&＃xff09;倾斜均衡配置 Hive自带了一个均衡数据倾斜的配置项。

其实现方法是在group by时启动两个MR job。第一个job会将map端数据随机输入reducer&＃xff0c;每个reducer做部分聚合&＃xff0c;相同的key就会分布在不同的reducer中。第二个job再将前面预处理过的数据按key聚合并输出结果&＃xff0c;这样就起到了均衡的效果。

hive.groupby.skewindata&＃61;false // 默认

1. join基础优化

&＃xff08;1&＃xff09; Hive在解析带join的SQL语句时&＃xff0c;会默认将最后一个表作为大表&＃xff0c;将前面的表作为小表&＃xff0c;将它们读进内存。如果表顺序写反&＃xff0c;如果大表在前面&＃xff0c;引发OOM。不过现在hive自带优化。

&＃xff08;2&＃xff09; map join:特别适合大小表join的情况&＃xff0c;大小表join在map端直接完成join过程&＃xff0c;没有reduce&＃xff0c;效率很高。

&＃xff08;3&＃xff09;多表join时key相同&＃xff1a;会将多个join合并为一个MR job来处理&＃xff0c;两个join的条件不相同&＃xff0c;就会拆成多个MR job计算。

1. sort by代替order by

将结果按某字段全局排序&＃xff0c;这会导致所有map端数据都进入一个reducer中&＃xff0c;在数据量大时可能会长时间计算不完。使用sort by&＃xff0c;那么还是会视情况启动多个reducer进行排序&＃xff0c;并且保证每个reducer内局部有序。为了控制map端数据分配到reducer的key&＃xff0c;往往还要配合distribute by一同使用。如果不加distribute by的话&＃xff0c;map端数据就会随机分配到reducer。

1. 单独处理倾斜key

一般来讲倾斜的key都很少&＃xff0c;我们可以将它们抽样出来&＃xff0c;对应的行单独存入临时表中&＃xff0c;然后打上随机数前缀&＃xff0c;最后再进行聚合。或者是先对key做一层hash&＃xff0c;先将数据随机打散让它的并行度变大&＃xff0c;再汇集。其实办法一样。