Hive架构简介
下图是hive的架构图
命令执行
所有的命令和查询都会进入到Driver，通过这个模块进行解析编译，对需求的计算进行优化。然后按照指定的步骤执行（通常是启动多个MapReduce任务（JOB）来执行）。当需要启动 MapReduce任务(job)时，HIVE 本身不会生成Java MapReduve算法程序。相反，Hive通过一个表示“JOB执行计划”的。XML文件驱动执行内置的、原生的Mapper和Reducer模块。换句话说，这些通用的模块函数类似于微型的语言翻译程序，二驱动计算的“语言”是以XML形式编码的。Hive 通过和JobTracker通信来初始化MapReduve任务(job)，而不必部署在JobTracker所在的管理节点上执行。要处理的数据文件是存储在HDFS中的，而HDFS是由NameNode进行管理的。
驱动器：Dirver 包含：解析器、编译器、优化器、执行器
解析器：将SQL字符串转换成抽象语法树AST，这一步一般都是用第三方工具库完成，比如antlr；对AST语 法树进行分析，比如表否存在、字段是否存在、SQL语义是否有误。
编译器：将AST编译生成逻辑执行计划。
优化器：对逻辑执行计划进行优化。
执行器：把逻辑执行计划转换成物理执行计划。对于hive来说，就是MR/TEZ/Spark；
元数据
Metastore是一个独立的关系型数据库，通常使用MYSQL。Hive 会在其中保存表模式和其他系统元数据中的。元数据包括表的名字，表的列和分区及其属性，表的属性（是否为外部表等），表的数据所在目录等
Hive数据类型
hive基础数据类型
hive是用java开发的，hive里的基本数据类型和java的基本数据类型也是一一对应的，除了string类型。有符号的整数类型：TINYINT、SMALLINT、INT和BIGINT分别等价于java的byte、short、int和long原子类型，它们分别为1字节、2字节、4字节和8字节有符号整数。Hive的浮点数据类型FLOAT和DOUBLE,对应于java的基本类型float和double类型。而hive的BOOLEAN类型相当于java的基本数据类型boolean。
hive复杂数据类型
包括ARRAY,MAP,STRUCT,UNION。这些复杂类型是由基础类型组成的。
ARRAY：ARRAY类型是由一系列同样数据类型元素组成的，这些元素能够通过下标来訪问。比方有一个ARRAY类型的变量fruits。它是由[‘apple’,’orange’,’mango’]组成，那么能够通过fruits[1]来訪问orange。
MAP：MAP包括key->value键值对。能够通过key来訪问元素。比方userlist是一个map类型（当中username是key。password是value），那么我们能够通过userlist[username]来得到这个用户相应的password.。
STRUCT：STRUCT能够包括不同数据类型的元素。这些元素能够通过点的方式来得到，比方user是一个STRUCT类型，那么能够通过user.address得到这个用户的地址。
COLLECTION ITEMS TERMINATED BY ',' //集合元素分隔符
MAP KEYS TERMINATED BY ' : ' //MAP数据类型key:value分割符
Hive表类型
内部表
内部表也称之为MANAGED_TABLE；默认存储在/user/hive/warehouse下，也可以通过location指定；删除表时，会删除表数据以及元数据；
create table if not exists...
外部表
外部表称之为EXTERNAL_TABLE在创建表时可以自己指定目录位置(LOCATION)；删除表时，只会删除元数据不会删除表数据；
create EXTERNAL table if not exists...
加载数据
注意 导入的路径必须是hive用户有访问权限的路径,否则会报找不到文件错误
(1) HDFS上导入数据到Hive表
load data inpath '/home/wyp/add.txt' into table wyp;
(2) 从本地路径导入数据到Hive表
load data local inpath 'wyp.txt' into table test ;
#使用分区
load data local inpath '/your/path/test.txt' overwrite into table test partition (age='25')
(3) 从别的表查询加载到hive表
静态分区
insert into table test partition (age='25') select id, name, tel from wyp;
动态分区
set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
insert overwrite table test PARTITION (age) select id, name, tel, age from wyp;
insert into table和insert overwrite table的区别：后者会覆盖相应数据目录下的数据将。
创建相似表
create table table_name like other_table_name location "xxxxx"
查询数据输出到本地目录
INSERT OVERWRITE DIRECTORY '/tmp/hdfs_out' SELECT a.* FROM invites a WHERE a.ds='<DATE>';
select *from emp distribute by deptno sort by empno asc ;
cluster by = distribute by 和 sort by 值相同。
Hive UDF编程
UDF(user defined function).hive的UDF包含三种：UDF支持一个输入产生一个输出，UDTF支持一个输入多个输出输出(一行变多行)，UDAF支持多输入一输出(多行变一行)。
UDF 简单实现
编程步骤：
1、继承org.apache.hadoop.hive.ql.UDF
2、需要实现evaluate函数； evaluate函数支持重载；
注意事项：
1、 UDF必须要有返回类型，可以返回null,但是返回类型不能为void;
2、 UDF中常用Text/LongWritable等类型，不推荐使用java类型;
官网demo: https://cwiki.apache.org/confluence/display/Hive/HivePlugins
使用UDF非常简单，只需要继承org.apache.hadoop.hive.ql.exec.UDF，并定义public Object evaluate(Object args) {} 方法即可。
比如，下面的UDF函数实现了对一个String类型的字符串取HashMD5：
package com.lxw1234.hive.udf;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.MD5Hash;
import org.apache.hadoop.hive.ql.exec.UDF;
public class HashMd5 extends UDF {
public String evaluate(String cookie) {
return MD5Hash.getMD5AsHex(Bytes.toBytes(cookie));
}
public String evaluate(String cookie1,String cookie2) {
return MD5Hash.getMD5AsHex(Bytes.toBytes(cookie1,cookie2));
}
}
将上面的HashMd5类打成jar包，udf.jar
使用时候，在Hive命令行执行：
hive> add jar file:///tmp/udf.jar;
hive> CREATE temporary function str_md5 as 'com.lxw1234.hive.udf.HashMd5';
hive> select str_md5(‘lxw1234.com’) from dual;
Transform
Hive中的TRANSFORM：使用脚本完成Map/Reduce。
hive> add file python文件路径
hive> SELECT
TRANSFORM(p.joint_attr_values, p.collect_product_id, p.released_id)
USING 'python split_product_attrs.py'
as (custom_attr , custom_attr_value, collect_product_id, released_product_id)
from
( 这里应该是另外一个select语句，用于Transform的输入，最好是一一对应的，否则会出错 )
注意：python脚本一定要先加载：hive>ADD FILE python_file_path [注意此处的path不要加''上引号]
下面是python的脚本，用于将三列转换为四列，这里就比较简单了，主要用于测试，代码随便写了一下
#!/usr/bin/python
# #_*_ coding: utf-8 _*_
import sys
import datetime
# "规格:RN1-10/50;规格:RN1-10/50;规格:RN1-10/50"
# ["规格:RN1-10/51;规格:RN1-10/52;规格:RN1-10/53", '11', '22']
# ["规格", "RN1-10/51", '11', '22']
# ["规格", "RN1-10/52", '11', '22']
# ["规格", "RN1-10/53", '11', '22']
for line in sys.stdin:
values = line.split('\t')
values = [ i.strip() for i in values ]
tmp = values[0]
key_values = tmp.split(";")
for kv in key_values:
k = kv.split(":")[0]
v = kv.split(":")[1]
print '\t'.join([k,v,values[1],values[2]])
Hive企业优化方案
Hive存储格式
1、TEXTFILE 默认格式，建表时不指定默认为这个格式，导入数据时会直接把数据文件拷贝到hdfs上不进行处理。只有TEXTFILE表能直接加载数据，本地load数据，和external外部表直接加载运路径数据，都只能用TEXTFILE表。更深一步，hive默认支持的压缩文件（hadoop默认支持的压缩格式），也只能用TEXTFILE表直接读取。其他格式不行。可以通过TEXTFILE表加载后insert到其他表中。
2、orc 格式。 STORED AS ORC ;
3、parquet 格式。 STORED AS PARQUET ;
几种格式的差别 http://www.cnblogs.com/juncaoit/p/6067646.html
Hive压缩优化
hive压缩相关博客 https://yq.aliyun.com/articles/60859
压缩配置：
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;
中间压缩
中间压缩就是处理作业map任务和reduce任务之间的数据，对于中间压缩，最好选择一个节省CPU耗时的压缩方式
<property>
<name>hive.exec.compress.intermediate</name>
<value>true</value>
</property>
hadoop压缩有一个默认的压缩格式，当然可以通过修改mapred.map.output.compression.codec属性，使用新的压缩格式，这个变量可以在mapred-site.xml 中设置或者在 hive-site.xml文件。 SnappyCodec 是一个较好的压缩格式，CPU消耗较低。
<property>
<name>mapred.map.output.compression.codec</name>
<value>org.apache.hadoop.io.compress.SnappyCodec</value>
</property>
MapReduce中使用压缩
Hive中 使用压缩
关闭推测执行
(1) 修改mapred-site.xml
mapreduce.map.speculative false
mapreduce.reduce.speculative false
(2) 设置值
set hive.mapred.reduce.tasks.speculative.execution=false;
(3) set hive.exec.mode.local.auto=true 决定 Hive 是否应该自动地根据输入文件大小，在本地运行。
Hive 数据倾斜解决方案
Hive join解释 https://www.cnblogs.com/xing901022/p/5804836.html
小表 join 大表 大表 join 大表 方案
（1）Common/Shuffle/Reduce Join
普通mapreduce join ,相同key分配到同一个reducer。
（2） Map Join
MAPJION会把小表全部读入内存中，在map阶段直接拿另外一个表的数据和内存中表数据做匹配,适用于小表join 大表。
set hive.auto.convert.join=false;
set hive.ignore.mapjoin.hint=false;
Sql可以通过使用hint的方式指定join时使用mapjoin，其中mapjoin选中的表为小表
select /*+ mapjoin(t1)*/ t1.a,t1.b from table t1 join table2 t2 on ( t1.a=t2.a and t1.time=201108)
Map Join Java实现 http://www.cnblogs.com/ivanny/p/mapreduce_join.html
（3） SMB Join （Sort Merge Bucket Join）
使用SMB两个表必须使用桶分区，使用 SMB Join 需要如下配置
set hive.enforce.bucketing = true; //强制使用桶表
set hive.auto.convert.sortmerge.join=true;
set hive.optimize.bucketmapjoin = true;
set hive.optimize.bucketmapjoin.sortedmerge = true;
select /*+ mapjoin(t1)*/ t1.a,t1.b from table t1 join table2 t2 on ( t1.a=t2.a and t1.time=201108)
- 两个表关联键为id，需要按id分桶并且做排序，小表的分桶数是大表分桶数的倍数。
2. 对于map端连接的情况，两个表以相同方式划分桶。处理左边表内某个桶的 mapper知道右边表内相匹配的行在对应的桶内。因此，mapper只需要获取那个桶 (这只是右边表内存储数据的一小部分)即可进行连接。这一优化方法并不一定要求 两个表必须桶的个数相同，两个表的桶个数是倍数关系也可以。
3. 桶中的数据可以根据一个或多个列另外进行排序。由于这样对每个桶的连接变成了高效的归并排序(merge-sort), 因此可以进一步提升map端连接的效率。
Hive分区
参数设置
#开启动态分区
set hive.exec.hynamic.partition=true;
#注：这个属性默认是strict，即限制模式，strict是避免全分区字段是动态的，必须至少一个分区字段是指定有值即静态的，且必须放在最前面。设置为nonstrict之后所有的分区都可以是动态的了。
set hive.exec.dynamic.partition.mode=nonstrict;
#注：这个属性表示每个节点生成动态分区的最大个数，默认是100
SET hive.exec.max.dynamic.partitions=100000;
#注：这个属性表示每个节点生成动态分区的最大个数，默认是100
SET hive.exec.max.dynamic.partitions.pernode=100000;
#注：这个属性表示一个DML操作可以创建的最大文件数，默认是100000
set hive.exec.max.created.files=150000
建表
create table user_info (user_id int, cid string, ckid string, username string)
row format delimited
fields terminated by '\t'
lines terminated by '\n';
1.行格式化
2.字段通过'\t' 分割
3.行通过'\n' 分隔开
# flume 收集 建表方案
create table baoertest(user_agent string,user_id bigint,time string)
partitioned by (year string, month string, day string)
clustered by (user_id) into 5 buckets
stored as orc
tblproperties("transactional"="true"); #使用这个，不能使用insert overwrite，可以使用insert,delete
# insert overwrite table 建表方案 prestodb可识别
create table baoertest(user_agent string,user_id bigint,time string)
partitioned by (year string, month string, day string)
clustered by (user_id) into 5 buckets
stored as orc;
set hive.enforce.bucketing=true; # 使用桶，需要强制设置
set mapreduce.job.reduces=
#静态分区
create table baoertest_txt(user_agent string,user_id bigint,time string)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';
LOAD DATA LOCAL INPATH '/data/hive/data.txt' OVERWRITE INTO TABLE baoertest_txt;
INSERT OVERWRITE TABLE baoertest PARTITION (year="2015",month="10",day="12")
select * from baoertest_txt;
--data.txt--
chrome 112212123 11-12
windows 123123123 11-11
#动态分区
set hive.exec.dynamic.partition=true; #动态分区
set hive.exec.dynamic.partition.mode=nonstrict; #动态分区非严格模式
create table baoertest_txt01(user_agent string,user_id bigint,time string,year string, month string, day string)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t';
LOAD DATA LOCAL INPATH '/data/hive/data01.txt' OVERWRITE INTO TABLE baoertest_txt01;
#默认查询出来的后三个字段作为partition字段
INSERT OVERWRITE TABLE baoertest PARTITION (year,month,day)
select * from baoertest_txt01;
--data01.txt--
chrome 112212123 11-12 2017 05 21
windows 123123123 11-11 2017 06 19
#查询使用分区
当分区建立后，查询可直接使用分区字段
select *from baoertest where year=xxx and month=xxx and day=xxx
#手动添加分区（ 外部表需要手动添加分区 ）
alter table table_xxx add partition (dt='2016-09-12');
#手动删除分区（ 删除分区 非外部表会同时删除分区的HDFS数据 ）
alter table table_xxx drop partition (month=2017,month=10,day=21);
#自动添加分区,根据HDFS目录名
MSCK REPAIR TABLE table_name
执行后，Hive会检测如果HDFS目录下存在但表的metastore中不存在的partition元信息，更新到metastore中
Hive 修改表
#修改表的location url
alter table test set location 'hdfs://heracles/user/video-mvc/hive/warehouse/t_m_cc'
