Hive入门函数提升

Posted 2021-05-18 杀智勇双全杀

Hive入门（六）函数提升

• 窗口聚合函数
• 窗口位置函数
• • first_value
  • last_value
  • lag
  • lead
• 窗口分析函数
• • row_number
  • rank
  • dense_rank
  • ntil
• Hive优化——参数优化
• • MapReduce参数优化
  • • 推测执行
    • JVM重用
  • Hive参数优化
  • • Fetch Task
    • 严格模式
    • 并行执行
    • 压缩
• Hive优化——SQL优化
• Hive优化——表设计优化
• • 分区表
  • 分桶表
  • 文件格式
  • • 列式存储
• Hive数据倾斜
• • 现象
  • 原因
  • 解决方案
  • • group by / count(distinct)中
    • • 开启Combiner
      • 随机分区
    • join中

Hive函数入门

窗口聚合函数

语法格式：

funName(参数) over (partition by col [order by col]  [window_szie])

其中：
函数可以是：max/min/avg/count/sum。

partition by：分区，将相同分区的数据放在一起。
order by：分区内部按照某个字段进行排序。
window_szie：窗口大小，指定的是函数处理数据的范围。如果指定了partition by和order by，没有指定窗口大小，默认窗口就是分区的第一行到当前行。指定窗口大小可以缩小范围，例如：
rows between 3 preceding and current row，就是该行上移3行到当前行的窗口。
rows between unbounded preceding and current row，就是从分区内起始行到当前行的窗口。rows between 3 preceding and 1 following，就是分区内，该行前移3行到后移1行的窗口。 between current row and unbounded following，就是从当前行到最后一行的窗体。

指定窗口大小有这些参数：
N preceding ：前N行。
N following ：后N行。
current row：当前行。
unbounded preceding 表示从前面的起点，第一行。
unbounded following：表示到后面的终点，最后一行。

窗口位置函数

first_value

取每个分区内某列的第一个值。

first_value(col) over (partition by col1 order by col2)。

last_value

取每个分区内某列的最后一个值。

last_value() over (partition by col1 order by col2)。

lag

取每个分区内某列的前N个值。

lag(col,N,defaultValue) over (partition by col1 order by col2)

其中：
col：取分区内某一列的值。
N：向前偏移N个单位。
defaultValue：如果取不到的默认值。

lead

取每个分区内某列的后N个值。

lead(col,N,defaultValue) over (partition by col1 order by col2)。

其中：
col：取分区内某一列的值。
N：向后偏移N个单位。
defaultValue：如果取不到的默认值。

窗口分析函数

row_number

用于实现分区内记录编号。

row_number() over (partition by col1 order by col2)。

如果值相同，继续编号。一定是1，2，3，4…这种情况。

rank

用于实现分区内排名编号，会留空位。

rank() over (partition by col1 order by col2)。

如果值相同，编号相同，会留下空位。如：1，1，3，4这种情况。

dense_rank

用于实现分区内排名编号，不留空位。

dense_rank() over (partition by col1 order by col2)。

如果值相同，编号相同，不留空位。如：1，1，2，3这种情况。

ntil

将每个分区内排序后的结果均分成N份，如果不能均分，优先分配编号小的。

ntil(N) over (partition by col1 order by col2)。

Hive优化——参数优化

MapReduce参数优化

推测执行

经常运行一个Mapreduce程序，有多个MapTask和ReduceTask，由于网络或者资源故障导致有一个Task一直不能运行结束。这种情况需要使用推测执行。

如果appmaster发现某个Task一直不能结束，会在另外节点上启动同一个Task，谁先运行结束，另外一个会被kill。

需要设置：

mapreduce.map.speculative=true
mapreduce.reduce.speculative=true
hive.mapred.reduce.tasks.speculative.execution=true

但是这种情况可能是由于数据倾斜导致的，如果因为资源不足卡死而不是断网，可能会因为资源抢占导致更严重的卡死。故这种方式只适合于资源充足的情况。

JVM重用

每次每个Task都会申请一个JVM进程来运行程序，JVM进程需要内存等资源，每个Task运行完成以后，这个JVM就被销毁了。重复的资源调度会浪费很多时间。

利用JVM重用，申请了一个JVM进程的资源以后，可以运行多个Task，实现资源的复用。

需要设置：

set  mapreduce.job.jvm.numtasks=10

Hive参数优化

Fetch Task

Hive自带了小型计算引擎，一部分简单的SQL语句不走Mapreduce，直接由Fetch Task处理。执行小任务时比MapReduce分布式计算快。

hive.fetch.task.conversion


Expects one of [none, minimal, more].
      Some select queries can be converted to single FETCH task minimizing latency.
      Currently the query should be single sourced not having any subquery and should not have
      any aggregations or distincts (which incurs RS), lateral views and joins.
      0. none : disable hive.fetch.task.conversion
      1. minimal : SELECT STAR, FILTER on partition columns, LIMIT only
      2. more    : SELECT, FILTER, LIMIT only (support TABLESAMPLE and virtual columns)

none：不启用fetch抓取。

严格模式

老版本的属性：hive.mapred.mode=nonstrict/strict。

如果为严格模式：hive会限制一些SQL语句的运行。

新版本的属性：hive.strict.checks.type.safe。

是否允许一些风险性的类型比较：
bigints and strings，
bigints and doubles。

hive.strict.checks.cartesian.product：是否允许笛卡尔的产生。

并行执行

Hive在解析SQL时，默认不会并行执行Stage，只会单个Stage执行。

设置并行度，可以提高Hive解析编译的性能。

set hive.exec.parallel=true;              //打开任务并行执行
set hive.exec.parallel.thread.number=16;  //同一个sql允许最大并行度，默认为8。

压缩

先配置Hadoop中的压缩，再设置Mapreduce的shuffle的中间输出压缩。

#配置多个Mapreduce中的中间Mapreduce的结果压缩
hive.exec.compress.intermediate=true

Hive优化——SQL优化

最有效的就是PPD：
谓词下推 Predicate Pushdown（PPD）的思想简单点说就是在不影响最终结果的情况下，尽量将过滤条件提前执行。谓词下推后，过滤条件在map端执行，减少了map端的输出，降低了数据在集群上传输的量，降低了Reduce端的数据负载，节约了集群的资源，也提升了任务的性能。

例如：where与having，能在where中过滤的就不要在having中过滤。

Hive优化——表设计优化

分区表

优化底层MapReduce输入，提高性能。

分桶表

提前将数据分桶存储，提高Join性能。

文件格式

创建表时可以指定

sorted as file_format

file_format:
  : SEQUENCEFILE
  | TEXTFILE    -- (Default, depending on hive.default.fileformat configuration)
  | RCFILE      -- (Note: Available in Hive 0.6.0 and later)
  | ORC         -- (Note: Available in Hive 0.11.0 and later)
  | PARQUET     -- (Note: Available in Hive 0.13.0 and later)
  | AVRO        -- (Note: Available in Hive 0.14.0 and later)
  | JSONFILE    -- (Note: Available in Hive 4.0.0 and later)
  | INPUTFORMAT input_format_classname OUTPUTFORMAT output_format_classname

默认格式：textFile。
二进制格式：SEQUENCEFILE。二进制文件没有编码，体积更大，但是IO读写更快。
列式存储：rcfile、orc、parquet。

列式存储

相同的数据，存储占用的空间更小。对于SQL分析读取列比较方便。

例如：

col1		col2		col3
1			a			c
2			b			d

按行存储：

1			a			c    2			b			d

按列存储：

1		2		a			b			c		d

使用：

select col1 from table

按行存储：将每一行都读取出来，然后过滤第一列。
按列存储：直接读取对应的列。

还可以构建列式索引。

Hive数据倾斜

现象

运行一个程序，这个程序的某一个Task一直在运行，其他的Task都运行结束了，进度卡在99%或者100%。

原因

基本原因：这个ReduceTask的负载要比其他Task的负载要高。ReduceTask的数据分配不均衡导致，MapTask（按照128M分片）不会有数据倾斜的问题。

根本原因：分区的规则。默认分区：根据K2的Hash值取余reduce的个数优点是相同的K2会由同一个reduce处理，缺点是可能导致数据倾斜。

解决方案

group by / count(distinct)中

开启Combiner

hive.map.aggr=true

随机分区

hive.groupby.skewindata=true

启动这个配置后，底层会自动走2个MapReduce。第一个MapReduce自动实现随机分区第二个MapReduce做最终的聚合。

或者手动指定：

distribute by rand()

但是这样会导致Map时相同K的数据不在同一个分区。

join中

尽量避免走Reduce Join。

Map Join中，尽量将不需要参加Join的数据过滤，将大表转换为小表。或者构建分桶Bucket Map Join。