Apache Hive高阶实战

Hive的事务（了解）：

一组单元化操做，这些操做要么都执行，要么都不执行，是一个不可分割的工做单位。

Atomicity（原子性）:

不可再分割的工做单位，事务中的全部操做要么都发生，要么都不发生。

Consistency（一致性）:

事务开始以前和事务结束之后，数据库的完整性约束没有被破坏。这是说数据库事务不能破坏关系数据的完整性以及业务逻辑上的一致性。

Isolation（独立性）:

多个事务并发访问，事务之间是隔离的，

Durability（持久性）:

意味着在事务完成之后，该事务锁对数据库所做的更改便持久的保存在数据库之中，并不会被回滚。

Hive事务的特色和局限：

从v0.14开始支持行级事务

支持INSERT，DELETE，UPDATE（从v2.2.0开始支持MERGE）

不支持BEGIN，COMMIT和ROLLBACK

不支持桶列或分区列的更新

文件格式只支持ORC

表必须是嵌套的群集表

须要压缩工做，这须要时间、资源和空间

锁能够是S（共享锁）或者X locks（排它锁）

X locks（排它锁）是串行的，而不是并发的

不容许从一个非ACID链接写入/读取ACID表

未彻底准备好生产

示例：

CREATE TABLE IF NOT EXISTS employee(

emp_id int,

emp_name string,

dept_name string,

work_loc string

)

PARTITIONED BY (start_date string)

CLUSTERED BY (emp_id) INTO 5 BUCKETS STORED AS ORC TBLPROPERTIES('transactional'='true');

INSERT INTO table employee PARTITION (start_date) VALUES

(1,'Will','IT','Toronto','20100701'),

(2,'Wyne','IT','Toronto','20100701'),

(3,'Judy','HR','Beijing','20100701'),

(4,'Lili','HR','Beijing','20101201'),

(5,'Mike','Sales','Beijing','20101201'),

(6,'Bang','Sales','Toronto','20101201'),

(7,'Wendy','Finance','Beijing','20101201');

 

Update employee set dept_name = ‘HR’ where emp_id = 7;

//正常运行

Update employee set emp_id = 7 where dept_name = ‘HR’;

//报错由于emp_id是分桶列

Update employee set start_date =’ ’ where dept_name = ‘HR’;

//报错由于start_date是分区列

MERGE INTO语句（V2.2以后）：

Merge容许基于与源表的链接的结果在目标表上执行操做。

Merge语句是用于数据仓库用例的很是强大的工具。它使SCD类型2的操做在Hive中更容易。

Merge只能在支持ACID的表上执行

标准的语法

MERGE INTO <target table> AS T USING <source expression/table> AS S ON

<boolean expression1>

WHEN MATCHED [AND <boolean expression2>] THEN UPDATE SET <set clause list>

WHEN MATCHED [AND <boolean expression3>] THEN DELETE

WHEN NOT MATCHED [AND <boolean expression4>] THEN INSERT VALUES<value list>

注意：

WHEN不能超过三个；每种类型最多1个：UPDATE/DELETE/INSERT

WHEN NOT MATCHED必须是最后一个WHEN子句。

若是同时存在UPDATE和DELETE子句，则语句中的第一个子句必须包含[AND <boolean

expression>]

Hive事务的开启和设置：

在命令行中，为当前会话设置：

    set hive.support.concurrency = true;

    set hive.enforce.bucketing = true;

    //必定要作的

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

    set hive.txn.manager = org.apache.hadoop.hive.ql.lockmgr.DbTxnManager;

    set hive.compactor.initiator.on = true;

    set hive.compactor.worker.threads = 1;

在hive-site.xml中设置：

    <property>

        <name>hive.support.concurrency</name>

        <value>true</value>

    </property>

    <property>

        <name>hive.txn.manager</name>

        <value>org.apache.hadoop.hive.ql.lockmgr.DbTxnManager</value>

    </property>

Hive中的UDF（User Defined Function）：

Java编写函数；继承UDF类，实现方法evaluate

1. org.apache.hadoop.hive.ql.exec.UDF
2. 参数和返回值类型位于org.apache.hadoop.io包

FloatWritable evaluate(final Text a){return t;}

打jar包

    maven依赖

    Hadoop.common,hive-exec

添加jar包到hive

    Add jar …./..jar（路径是Linux本地路径）

建立udf

    Create function

    salary_number as 类名；

用户定义函数(UDF)：这提供了一种使用外部函数(在Java中)扩展功能的方法，该外部函数能够在HQL中计算。

UDF：用户定义的函数，按行操做并为一行输出一个结果，例如大多数内置的数学和字符串函数。

UDAF：用户定义的聚合函数，按行或按组操做，结果为每一个组输出一行或一行，例如内置的MAX和COUNT函数。

UDTF：用户定义的表生成函数，也能够按行操做，可是它们会生成多个行/表，例如EXPLODE函数。UDTF能够在SELECT以后使用，也能够在LATERAL VIEW语句后使用。

Hive UDF开发步骤：

扩展适当的UDF类或GenericUDF类

重写适当的方法和实现功能逻辑

编译和构建 .jar文件

将jar文件复制到适当的HDFS路径

建立临时或永久性功能jar

检查SHOW/DESC创造的功能

当你不须要的时候删除功能

UDF部署遍历

在shell命令行中，将jar文件上传到HDFS：

hdfs dfs -mkdir -p /apps/hive/functions

hdfs dfs -put -f df-hiveudf-1.0-SNAPSHOT.jar /apps/hive/functions

在beeline中，建立/注册函数：

DROP FUNCTION IF EXISTS str_lower;

DROP FUNCTION IF EXISTS array_contains;

CREATE FUNCTION str_lower AS 'com.datafibers.hiveudf.udf.StringLower'

ADD JAR /apps/hive/functions/df-hiveudf-1.0-SNAPSHOT.jar;

CREATE FUNCTION array_contains AS 'com.datafibers.hiveudf.gudf.ArrayContains'

ADD JAR /apps/hive/functions/df-hiveudf-1.0-SNAPSHOT.jar;

在那以后测试：

select str_lower(name), work_place from employee where array_contains(work_place, 'Toronto');

Hive中PLSQL概述

Hive PLSQL是Hive自v.2以来的一个过程语言接口

它还支持SparkSQL和Impala

兼容Oracle、DB二、MySQL、TSQL标准

它的一个目的是将现有的过程简单有效地迁移到Hadoop Hive

编写UDF不须要Java技能

它的性能比Java UDF稍慢

它仍然是新的

Hive PLSQL环境：

hplsql命令不在$PATH中，所以咱们须要直接在它的位置运行它

cd /opt/modules/hive-2.3.4/bin

Hive中PLSQL代码

CREATE FUNCTION hello(text STRING)

RETURNS STRING

BEGIN

RETURN 'Hello, ' || text || '!';

END;

Print hello(' word')

 

[root@sandbox-hdp bin]./hplsql -f plsql_demo.pl

Hello, word!

Hive性能调节：

性能效用：

EXPLAIN—不运行就能够返回查询执行计划

EXPLAIN [EXTENDED|DEPENDENCY|AUTHORIZATION] hive_query

EXTENDED：这为计划中的操做符提供了额外的信息，好比Map和Reduce端文件路径、统计信息和语法树

DEPENDENCY：这提供了JSON格式的输出，其中包含查询所依赖的表和分区的列表。

AUTHORIZATION：列出全部须要受权的实体，包括运行Hive查询和受权失败所需的输入和输出。

ANALYZE—收集统计信息，例如行数、文件数、最大值、最小值、均值，数据库中的表、分区、列等对象上的原始数据大小。这些统计信息做为基于hive成本优化器的输入，全部现代数据库都使用这种方法，使用ANALYZE以后统计出的数据会存在系统中，若是进行相关查询时系统就会直接调用，不须要使用Map-Reduce了。

ANALYZE TABLE employee COMPUTE STATISTICS;

ANALYZE TABLE employee_partitioned PARTITION(year=2014, month=12) COMPUTE STATISTICS;

ANALYZE TABLE employee_id COMPUTE STATISTICS FOR COLUMNS employee_id;

DESCRIBE EXTENDED table_name

//检查统计信息

设计优化：

1. 使用分区表（partition），最好提高性能的途径，加快查询速度，最经常使用的是使用时间进行partition
2. 使用桶表（bucket），减轻处理数据的压力，要清楚业务逻辑，熟悉主外键
3. 使用指数（index），有另一个表，对应主表的信息，在实际项目中用的很少
4. 使用适当的文件格式，而不是文本，如orc，avro，parquet；作查询，过滤时使用orc，parquet性能比较高
5. 使用适当的压缩，如snappy，能够减小空间，增长传输速度，可是压缩和解压缩也须要时间
6. 考虑数据本地化——增长一些复制因子，数据离咱们越近处理速度越快，对常用的数据进行拷贝
7. 避免小文件，小文件有两个来源：1. 元数据就是小文件；2. 开发人员操做不当，使用了错误的动态partition分区；小文件的影响：加大NameNode的运行压力，性能会下降，每个小文件都会占用128MB，若是小文件太多就会占用不少空间；解决小文件的方法：文件合并，Hive在运算时就有不少自动合并文件的功能，也能够写第三方的程序来合并小文件，数据源尽可能使用大数据
8. 使用Tez引擎代替mapreduce
9. 使用Hive LLAP（在内存读缓存中），比较像线程的想法，以前的Hive每运行一次就会启动一个新的线程，LLAP就在后台一直启动一个线程，会节省一个线程的启动
10. 考虑在不须要时关闭并发

做业优化—在本地模式下运行：

当要处理的数据很小时，启动分布式数据处理是一种开销，由于彻底分布式模式的启动时间比做业处理时间要长。Hive支持在本地模式下自动转换做业，设置以下：

直接使用set+想要查询的参数就能够查看这个参数如今设置的参数

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

    //默认值是false

    SET hive.exec.mode.local.auto.inputbytes.max=50000000;

    SET hive.exec.mode.local.auto.input.files.max=5;

    //默认值是4

做业必须知足如下条件才能在本地模式下运行：

         做业的总输入大小小于hive.exec.mode.local.auto. inputbytes.max

         映射任务的总数小于hive.exec.mode.local.auto. input.files.max

         reduce所需的任务总数为1或0

做业优化—JVM重用：

默认状况下，Hadoop为每一个map或reduce做业启动一个新的JVM，并并行运行map或reduce任务。当map或reduce做业是只运行几秒钟的轻量级做业时，JVM启动过程可能会带来很大的开销。

MapReduce框架（只有版本1，而不是Yarn）有一个重用JVM的选项，即共享JVM来串行而不是并行地运行mapper/reducer。JVM重用适用于映射或减小同一做业中的任务。来自不一样做业的任务老是在单独的JVM中运行。

    SET mapred.job.reuse.jvm.num.tasks=5;

    //默认值是1

做业优化—并行执行：

Hive查询一般被转换成许多按默认顺序执行的阶段。这些阶段并不老是相互依赖的。相反，它们能够并行运行，以节省整个做业运行时间。咱们能够经过如下设置启用此功能

    SET hive.exec.parallel=true;

    //默认值是false

    SET hive.exec.parallel.thread.number=16;

    //默认值是8，它定义并行运行的最大数量

并行执行将增长集群的利用率。若是集群的利用率已经很是高，那么并行执行对整体性能没有多大帮助。

查询优化：

确保自动启用map端join

确保桶映射侧链接是自动启用的

对于倾斜数据，设置hive.optimize.skewjoin=true

确保CBO是启用的（默认状况下是启用的）

    set hive.cbo.enable=true;

    set hive.compute.query.using.stats=true;

    set hive.stats.fetch.column.stats=true;

    set hive.stats.fetch.partition.stats=true;

确保启用了向量化。它改进了扫描、聚合、筛选和链接等操做，一次执行1024行批次，而不是每次执行一行

    set hive.vectorized.execution.enabled = true;

    set hive.vectorized.execution.reduce.enabled = true;

使用CTE、临时表、窗口函数等适当的编码规定

尽量使用CTE，避免使用嵌套子查询，还可使用临时表，省去了不少表；尽可能使用窗口函数，有优点，把查询写的十分简单，用简单的操做进行复杂的计算，有效的增长性能。

Hive on MapReduce调优，一般须要注意哪些？

合理设置job中reduce的数量和最大数量

set mapred.reduce.tasks

set hive.exec.reducers.max

合理设置job中每一个reduce处理文件的大小

set hive.exec.reducers.bytes.per.reducer

合理设置动态分区

set hive.exec.dynamic.partition

合理设置压缩

map/reduce 输出压缩（通常采用序列化文件存储）

set hive.exec.compress.output=true;

set mapred.output.compression.codec=org.apache.hadoop.io.compress.GzipCodec;

set mapred.output.compression.type=BLOCK;

任务中间压缩

set hive.exec.compress.intermediate=true;

set hive.intermediate.compression.codec=org.apache.hadoop.io.compress.SnappyCodec;（经常使用）

set hive.intermediate.compression.type=BLOCK;

合理设置输入分片

set mapreduce.input.fileinputformat.split.minsize //启动map最小的split size大小，默认0

set mapreduce.input.fileinputformat.split.maxsize //启动map最大的split size大小，默认256M

合理设置合并

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

set hive.merge.mapFiles=true;

set hive.merge.mapredFiles=true;

set hive.merge.size.per.task=xxxx