大数据 - （三） - Hadoop生态圈技术栈 - Hive

2020-08-10  本文已影响0人  啦啦啦喽啰

数据仓库工具 - Hive

Hive 产生背景

直接使用MapReduce处理海量数据，面临以下问题：

Hive基于Hadoop的一个数据仓库工具，可以将结构化的数据文件映射成一张表，并提供SQL查询功能

Hive本质

什么是数据仓库

数据仓库（Data Warehouse）是一个，面向主题、集成的、相对稳定、反应历史变化的数据集合，主要用于管理决策

Hive与RDBMS（关系型数据库管理系统）对比

Hive的优缺点

优点

缺点

Hive架构

image.png

Hive安装配置[^xr1]

版本
# hive安装包 apache-hive-2.3.7-bin.tar.gz
# MySQL安装包 mysql-5.7.26-1.el7.x86_64.rpm-bundle.tar
# MySQL的JDBC驱动程序 mysql-connector-java-5.1.46.jar
MySQL安装
# 查询是否安装了
mariadb rpm -aq | grep mariadb
# 删除mariadb。-e 删除指定的套件;--nodeps 不验证套件的相互关联性 
rpm -e --nodeps mariadb-libs

yum install perl -y
yum install net-tools -y

# 解压缩
tar xvf mysql-5.7.26-1.el7.x86_64.rpm-bundle.tar
# 依次运行以下命令
rpm -ivh mysql-community-common-5.7.26-1.el7.x86_64.rpm
rpm -ivh mysql-community-libs-5.7.26-1.el7.x86_64.rpm
rpm -ivh mysql-community-client-5.7.26-1.el7.x86_64.rpm
rpm -ivh mysql-community-server-5.7.26-1.el7.x86_64.rpm


systemctl start mysqld

grep password /var/log/mysqld.log

# 进入MySQL，使用前面查询到的口令 mysql -u root -p
# 设置口令强度;将root口令设置为12345678;刷新
set global validate_password_policy=0;
set password for 'root'@'localhost' =password('12345678'); flush privileges;

-- 创建用户设置口令、授权、刷新
CREATE USER 'hive'@'%' IDENTIFIED BY '12345678'; GRANT ALL ON *.* TO 'hive'@'%';
FLUSH PRIVILEGES;
hive安装
tar zxvf apache-hive-2.3.7-bin.tar.gz -C /opt/servers/
mv apache-hive-2.3.7-bin hive-2.3.7
# 在 /etc/profile 文件中增加环境变量
export HIVE_HOME=/opt/servers/hive-2.3.7
export PATH=$PATH:$HIVE_HOME/bin

<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
  <!-- hive元数据的存储位置 -->
  <property>
    <name>javax.jdo.option.ConnectionURL</name>
    <value>jdbc:mysql://os3:3306/hivemetadata?createDatabaseIfNotExist=true&amp;useSSL=false</value>
    <description>JDBC connect string for a JDBC metastore</description>
  </property>
  <!-- 指定驱动程序 -->
  <property>
    <name>javax.jdo.option.ConnectionDriverName</name>
    <value>com.mysql.jdbc.Driver</value>
    <description>Driver class name for a JDBC metastore</description>
  </property>
  <!-- 连接数据库的用户名 -->
  <property>
    <name>javax.jdo.option.ConnectionUserName</name>
    <value>hive</value>
    <description>username to use against metastore database</description>
  </property>
  <!-- 连接数据库的口令 -->
  <property>
   <name>javax.jdo.option.ConnectionPassword</name>
   <value>12345678</value>
   <description>password to use against metastore database</description>
  </property>
</configuration>

cp mysql-connector-java-5.1.46.jar $HIVE_HOME/lib

 schematool -dbType mysql -initSchema

hive
Hive 属性配置
<property>
  <!-- 数据默认的存储位置(HDFS) --> 
  <name>hive.metastore.warehouse.dir</name> 
  <value>/user/hive/warehouse</value> 
  <description>location of default database for the warehouse</description>
</property>

<property>
  <!-- 在命令行中，显示当前操作的数据库 -->
  <name>hive.cli.print.current.db</name>
  <value>true</value>
  <description>Whether to include the current database in the Hive prompt.</description> 
</property>

<property>
  <!-- 在命令行中，显示数据的表头 --> 
  <name>hive.cli.print.header</name> 
  <value>true</value>
</property>

<property>
  <!-- 操作小规模数据时，使用本地模式，提高效率 --> 
  <name>hive.exec.mode.local.auto</name>
  <value>true</value>
  <description>Let Hive determine whether to run in local mode automatically</description>
</property>
  • job的输入数据量必须小于参数：hive.exec.mode.local.auto.inputbytes.max (默认128MB)
  • job的map数必须小于参数：hive.exec.mode.local.auto.tasks.max (默认4)
  • job的reduce数必须为0或者1

参数配置方式

-- 查看全部参数 hive> set;
-- 查看某个参数
hive> set hive.exec.mode.local.auto; 
``
* 参数配置的三种方式:

1、用户自定义配置文件(hive-site.xml)
2、启动hive时指定参数(-hiveconf)
3、hive命令行指定参数(set)
配置信息的优先级:

set > -hiveconf > hive-site.xml > hive-default.xml

Hive命令

hive -e "select * from users"

# 执行文件中的SQL语句
hive -f hqlfile1.sql
# 执行文件中的SQL语句，将结果写入文件
hive -f hqlfile1.sql >> result1.log

数据类型与文件格式

基本数据类型及转换

整型
浮点数
定点数
字符串
布尔型
二进制类

数据类型的隐式转换

hive> select '1.0'+2;
OK
3.0
hive> select '1111' > 10;
hive> select 1 > 0.8;

数据类型的显示转换

hive> select cast('1111s' as int);
OK
NULL
hive> select cast('1111' as int);
OK
1111

集合数据类型

文本文件数据编码

# id name age hobby(array) score(map)
# 字段之间：^A
# 元素之间: ^B
# key-value之间：^C
# 666^Alisi^A18^Aread^Bgame^Ajava^C97^Bhadoop^C87
create table s1(
 id int,
 name string,
 age int,
 hobby array<string>,
 score map<string, int>
);
load data local inpath '/home/hadoop/data/s1.dat' into table s1;
select * from s1;

读时模式

"写时模式"（schema on write）

在传统数据库中，在加载时发现数据不符合表的定义，则拒绝加载数据。数据在写入数据库时对照表模式进行检查。

"读时模式"（schema on read）

加载数据时不进行数据格式的校验，读取数据时如果不合法则显示NULL。这种模式的优点是加载数据迅速。

HQL操作之 -- DDL命令

数据库操作

创建数据库语法
CREATE DATABASE [IF NOT EXISTS] database_name
[COMMENT database_comment]
[LOCATION hdfs_path]
[MANAGEDLOCATION hdfs_path]
[WITH DBPROPERTIES (property_name=property_value, ...)];

使用

-- 创建数据库，在HDFS上存储路径为 /user/hive/warehouse/*.db
create database if not exists mydb;
dfs -ls /user/hive/warehouse;
-- 指定位置
create database if not exists mydb2
comment 'this is mydb2'
location '/user/hive/mydb2.db';
查看数据库
-- 查看所有数据库
show database;
-- 查看数据库信息
describe database extended mydb2;
使用数据库
use mydb;
删除数据库
-- 删除一个空数据库
drop database databasename;
-- 如果数据库不为空，使用 cascade 强制删除
drop database databasename cascade;
建表语法
create [external] table [IF NOT EXISTS] table_name
[(colName colType [comment 'comment'], ...)]
[comment table_comment]
[partition by (colName colType [comment col_comment], ...)]
[clustered BY (colName, colName, ...)
[sorted by (col_name [ASC|DESC], ...)] into num_buckets buckets]
[row format row_format]
[stored as file_format]
[LOCATION hdfs_path]
[TBLPROPERTIES (property_name=property_value, ...)]
[AS select_statement];

内部表 & 外部表

内部表
-- 创建内部表
create table t1(
id int,
name string,
hobby array<string>,
addr map<string, string>
)
row format delimited
fields terminated by ";"
collection items terminated by ","
map keys terminated by ":";
-- 显示表的定义，显示的信息较少
desc t1;
-- 显示表的定义，显示的信息多，格式友好
desc formatted t1;
-- 加载数据
load data local inpath '/home/hadoop/data/t1.dat' into table t1;
-- 查询数据
select * from t1;
-- 查询数据文件
dfs -ls /user/hive/warehouse/mydb.db/t1;
-- 删除表。表和数据同时被删除
drop table t1;
-- 再次查询数据文件，已经被删除
外部表
-- 创建外部表
create external table t2(
id int,
name string,
hobby array<string>,
addr map<string, string>
)
row format delimited
fields terminated by ";"
collection items terminated by ","
map keys terminated by ":";
-- 显示表的定义
desc formatted t2;
-- 加载数据
load data local inpath '/home/hadoop/data/t1.dat' into table t2;
-- 查询数据
select * from t2;
-- 删除表。表删除了，目录仍然存在
drop table t2;
-- 再次查询数据文件，仍然存在
内部表与外部表的转换
-- 创建内部表，加载数据，并检查数据文件和表的定义
create table t1(
id int,
name string,
hobby array<string>,
addr map<string, string>
)
row format delimited
fields terminated by ";"
collection items terminated by ","
map keys terminated by ":";
load data local inpath '/home/hadoop/data/t1.dat' into table t1;
dfs -ls /user/hive/warehouse/mydb.db/t1;
desc formatted t1;
-- 内部表转外部表
alter table t1 set tblproperties('EXTERNAL'='TRUE');
-- 查询表信息，是否转换成功
desc formatted t1;
-- 外部表转内部表。EXTERNAL 大写，false 不区分大小
alter table t1 set tblproperties('EXTERNAL'='FALSE');
-- 查询表信息，是否转换成功
desc formatted t1;

分区表

分区表创建与数据加载
-- 创建表
create table if not exists t3(
id int
,name string
,hobby array<string>
,addr map<String,string>
)
partitioned by (dt string)
row format delimited
fields terminated by ';'
collection items terminated by ','
map keys terminated by ':';
-- 加载数据。
load data local inpath "/data/hadoop/data/t1.dat" into table t3
partition(dt="2020-06-01");
load data local inpath "/home/hadoop/data/t1.dat" into table t3
partition(dt="2020-06-02");
查看分区
show partitions t3;
新增分区并设置数据
-- 增加一个分区，不加载数据
alter table t3 add partition(dt='2020-06-03');
-- 增加多个分区，不加载数据
alter table t3
add partition(dt='2020-06-05') partition(dt='2020-06-06');
-- 增加多个分区。准备数据
hdfs dfs -cp /user/hive/warehouse/mydb.db/t3/dt=2020-06-01
/user/hive/warehouse/mydb.db/t3/dt=2020-06-07
hdfs dfs -cp /user/hive/warehouse/mydb.db/t3/dt=2020-06-01
/user/hive/warehouse/mydb.db/t3/dt=2020-06-08
-- 增加多个分区。加载数据
alter table t3 add
partition(dt='2020-06-07') location
'/user/hive/warehouse/mydb.db/t3/dt=2020-06-07'
partition(dt='2020-06-08') location
'/user/hive/warehouse/mydb.db/t3/dt=2020-06-08';
-- 查询数据
select * from t3;
修改分区的hdfs路径
alter table t3 partition(dt='2020-06-01') set location
'/user/hive/warehouse/t3/dt=2020-06-03';
删除分区
-- 可以删除一个或多个分区，用逗号隔开
alter table t3 drop partition(dt='2020-06-03'),
partition(dt='2020-06-04');

分桶表

分桶的原理

-- 测试数据
1   java    90
1   c   78
1   python  91

-- 创建分桶表
create table course(
id int,
name string,
score int
)
clustered by (id) into 3 buckets
row format delimited fields terminated by "\t";
-- 创建普通表
create table course_common(
id int,
name string,
score int
)
row format delimited fields terminated by "\t";
-- 普通表加载数据
load data local inpath '/home/hadoop/data/course.dat' into table
course_common;
-- 通过 insert ... select ... 给桶表加载数据
insert into table course select * from course_common;
-- 观察分桶数据。数据按照：(分区字段.hashCode) % (分桶数) 进行分区
dfs -cat /user/hive/warehouse/mydb.db/course/000000_0
3       hadoop  91
3       python  89
3       c       74
3       java    98
6       c       76
-- 按id值%3进行分区

修改表 & 删除表

-- 修改表名。rename
alter table course_common
rename to course_common1;
-- 修改列名。change column
alter table course_common1
change column id cid int;
-- 修改字段类型。change column
alter table course_common1
change column cid cid string;
-- The following columns have types incompatible with the
existing columns in their respective positions
-- 修改字段数据类型时，要满足数据类型转换的要求。如int可以转为string，但是
string不能转为int
-- 增加字段。add columns
alter table course_common1
add columns (common string);
-- 删除字段：replace columns
-- 这里仅仅只是在元数据中删除了字段，并没有改动hdfs上的数据文件
alter table course_common1
replace columns(
id string, cname string, score int);
-- 删除表
drop table course_common1;

HQL操作之--数据操作

数据导入

装载数据(Load)
LOAD DATA [LOCAL] INPATH 'filepath'
[OVERWRITE] INTO TABLE tablename [PARTITION (partcol1=val1,
partcol2=val2 ...)]

-- 创建表
CREATE TABLE tabA (
id int
,name string
,area string
) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' ;
数据文件（~/data/sourceA.txt）：
1,fish1,SZ
2,fish2,SH
3,fish3,HZ
4,fish4,QD
5,fish5,SR
-- 拷贝文件到 HDFS
hdfs dfs -put sourceA.txt data/
-- 装载数据：
-- 加载本地文件到hive(tabA)
LOAD DATA LOCAL INPATH '/home/hadoop/data/sourceA.txt'
INTO TABLE tabA;
-- 检查本地文件还在
-- 加载hdfs文件到hive(tabA)
LOAD DATA INPATH 'data/sourceA.txt'
INTO TABLE tabA;
-- 检查HDFS文件，已经被转移
-- 加载数据覆盖表中已有数据
LOAD DATA INPATH 'data/sourceA.txt'
OVERWRITE INTO TABLE tabA;
-- 创建表时加载数据
hdfs dfs -mkdir /user/hive/tabB
hdfs dfs -put sourceA.txt /user/hive/tabB
CREATE TABLE tabB (
id INT
,name string
,area string
) ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
Location '/user/hive/tabB';
插入数据(Insert)
-- 创建分区表
CREATE TABLE tabC (
id INT
,name string
,area string
)
partitioned by (month string)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';
-- 插入数据
insert into table tabC
partition(month='202001')
values (5, 'wangwu', 'BJ'), (4, 'lishi', 'SH'), (3, 'zhangsan',
'TJ');
-- 插入查询的结果数据
insert into table tabC partition(month='202002')
select id, name, area from tabC where month='202001';
-- 多表（多分区）插入模式
from tabC
insert overwrite table tabC partition(month='202003')
select id, name, area where month='202002'
insert overwrite table tabC partition(month='202004')
select id, name, area where month='202002';
创建表并插入数据(as select)
-- 根据查询结果创建表
create table if not exists tabD
as select * from tabC;
使用import导入数据
import table student2 partition(month='201709')
from '/user/hive/warehouse/export/student';

数据导出

-- 将查询结果导出到本地
insert overwrite local directory '/home/hadoop/data/tabC'
select * from tabC;
-- 将查询结果格式化输出到本地
insert overwrite local directory '/home/hadoop/data/tabC2'
row format delimited fields terminated by ' '
select * from tabC;
-- 将查询结果导出到HDFS
insert overwrite directory '/user/hadoop/data/tabC3'
row format delimited fields terminated by ' '
select * from tabC;
-- dfs 命令导出数据到本地。本质是执行数据文件的拷贝
dfs -get /user/hive/warehouse/mydb.db/tabc/month=202001
/home/hadoop/data/tabC4
-- hive 命令导出数据到本地。执行查询将查询结果重定向到文件
hive -e "select * from tabC" > a.log
-- export 导出数据到HDFS。使用export导出数据时，不仅有数还有表的元数据信息
export table tabC to '/user/hadoop/data/tabC4';
-- export 导出的数据，可以使用 import 命令导入到 Hive 表中
-- 使用 like tname创建的表结构与原表一致。create ... as select ... 结构
可能不一致
create table tabE like tabc;
import table tabE from ''/user/hadoop/data/tabC4';
-- 截断表，清空数据。(注意：仅能操作内部表)
truncate table tabE;
-- 以下语句报错，外部表不能执行 truncate 操作
alter table tabC set tblproperties("EXTERNAL"="TRUE");
truncate table tabC;

HQL操作之--DQL命令

DQL —— Data Query Language 数据查询语言
select语法

SELECT [ALL | DISTINCT] select_expr, select_expr, ...
FROM table_reference
[WHERE where_condition]
[GROUP BY col_list]
[ORDER BY col_list]
[CLUSTER BY col_list | [DISTRIBUTE BY col_list] [SORT BY
col_list]]
[LIMIT [offset,] rows]

基本查询

-- 省略from子句的查询
select 8*888 ;
select current_date ;
-- 使用列别名
select 8*888 product;
select current_date as currdate;
-- 全表查询
select * from emp;
-- 选择特定列查询
select ename, sal, comm from emp;
-- 使用函数
select count(*) from emp;
-- count(colname) 按字段进行count，不统计NULL
select sum(sal) from emp;
select max(sal) from emp;
select min(sal) from emp;
select avg(sal) from emp;
-- 使用limit子句限制返回的行数
select * from emp limit 3;

where子句

-- 使用 in
select * from emp where deptno in (20, 30);
-- 使用 between ... and ...
select * from emp where sal between 1000 and 2000;
-- 使用 like
select ename, sal from emp where ename like '%L%';
-- 使用 rlike。正则表达式，名字以A或S开头
select ename, sal from emp where ename rlike '^(A|S).*';

group by子句

-- 计算emp每个部门中每个岗位的最高薪水
select deptno, job, max(sal)
from emp
group by deptno, job;
group + having
-- 求每个部门的平均薪水大于2000的部门
select deptno, avg(sal)
from emp
group by deptno
having avg(sal) > 2000;

表连接

多表连接
select *
from techer t left join course c on t.t_id = c.t_id
left join score s on s.c_id = c.c_id
left join student stu on s.s_id = stu.s_id;
笛卡尔积
set hive.strict.checks.cartesian.product=false;
select * from u1, u2;

排序子句

全局排序(order by)
-- 别名排序
select empno, ename, job, mgr, sal + nvl(comm, 0) salcomm, deptno
from emp
order by salcomm desc;
-- 多列排序
select empno, ename, job, mgr, sal + nvl(comm, 0) salcomm, deptno
from emp
order by deptno, salcomm desc;
-- 排序字段要出现在select子句中。以下语句无法执行（因为select子句中缺少
deptno）：
select empno, ename, job, mgr, sal + nvl(comm, 0) salcomm
from emp
order by deptno, salcomm desc;

每个MR内部排序(sort by)

-- 设置reduce个数
set mapreduce.job.reduces=2;
-- 按照工资降序查看员工信息
select * from emp sort by sal desc;
-- 将查询结果导入到文件中（按照工资降序）。生成两个输出文件，每个文件内部数据按
工资降序排列
insert overwrite local directory '/home/hadoop/output/sortsal'
select * from emp sort by sal desc;

分区排序(distribute by)

-- 启动2个reducer task；先按 deptno 分区，在分区内按 sal+comm 排序
set mapreduce.job.reduces=2;
-- 将结果输出到文件，观察输出结果
insert overwrite local directory '/home/hadoop/output/distBy'
select empno, ename, job, deptno, sal + nvl(comm, 0) salcomm
from emp
distribute by deptno
sort by salcomm desc;
-- 上例中，数据被分到了统一区，看不出分区的结果
-- 将数据分到3个区中，每个分区都有数据
set mapreduce.job.reduces=3;
insert overwrite local directory '/home/hadoop/output/distBy1'
select empno, ename, job, deptno, sal + nvl(comm, 0) salcomm
from emp
distribute by deptno
sort by salcomm desc;

Cluster By

-- 语法上是等价的
select * from emp distribute by deptno sort by deptno;
select * from emp cluster by deptno;

函数

[Hive内置函数](https://cwiki.apache.org/confluence/display/Hive/LanguageMan
ual+UDF#LanguageManualUDF-Built-inFunctions)

系统内置函数

查看系统函数
-- 查看系统自带函数
show functions;
-- 显示自带函数的用法
desc function upper;
desc function extended upper;

日期函数

-- 当前前日期
select current_date;
select unix_timestamp();
-- 建议使用current_timestamp，有没有括号都可以
select current_timestamp();
-- 时间戳转日期
select from_unixtime(1505456567);
select from_unixtime(1505456567, 'yyyyMMdd');
select from_unixtime(1505456567, 'yyyy-MM-dd HH:mm:ss');
-- 日期转时间戳
select unix_timestamp('2019-09-15 14:23:00');
-- 计算时间差
select datediff('2020-04-18','2019-11-21');
select datediff('2019-11-21', '2020-04-18');
-- 查询当月第几天
select dayofmonth(current_date);
-- 计算月末:
select last_day(current_date);
-- 当月第1天:
select date_sub(current_date, dayofmonth(current_date)-1)
-- 下个月第1天:
select add_months(date_sub(current_date,
dayofmonth(current_date)-1), 1)
-- 字符串转时间（字符串必须为：yyyy-MM-dd格式）
select to_date('2020-01-01');
select to_date('2020-01-01 12:12:12');
-- 日期、时间戳、字符串类型格式化输出标准时间格式
select date_format(current_timestamp(), 'yyyy-MM-dd HH:mm:ss');
select date_format(current_date(), 'yyyyMMdd');
select date_format('2020-06-01', 'yyyy-MM-dd HH:mm:ss');
-- 计算emp表中，每个人的工龄
select *, round(datediff(current_date, hiredate)/365,1)
workingyears from emp;

字符串函数

-- 转小写。lower
select lower("HELLO WORLD");
-- 转大写。upper
select lower(ename), ename from emp;
-- 求字符串长度。length
select length(ename), ename from emp;
-- 字符串拼接。 concat / ||
select empno || " " ||ename idname from emp;
select concat(empno, " " ,ename) idname from emp;
-- 指定分隔符。concat_ws(separator, [string | array(string)]+)
SELECT concat_ws('.', 'www', array('test', 'com'));
select concat_ws(" ", ename, job) from emp;
-- 求子串。substr
SELECT substr('www.test.com', 5);
SELECT substr('www.test.com', -5);
SELECT substr('www.test.com', 5, 5);
-- 字符串切分。split，注意 '.' 要转义
select split("www.test.com", "\\.");

数学函数

-- 四舍五入。round
select round(314.15926);
select round(314.15926, 2);
select round(314.15926, -2);
-- 向上取整。ceil
select ceil(3.1415926);
-- 向下取整。floor
select floor(3.1415926);
-- 其他数学函数包括：绝对值、平方、开方、对数运算、三角运算等

条件函数

-- if (boolean testCondition, T valueTrue, T valueFalseOrNull)
select sal, if (sal<1500, 1, if (sal < 3000, 2, 3)) from emp;
-- CASE WHEN a THEN b [WHEN c THEN d]* [ELSE e] END
-- 将emp表的员工工资等级分类：0-1500、1500-3000、3000以上
select sal, if (sal<=1500, 1, if (sal <= 3000, 2, 3)) from emp;
-- CASE WHEN a THEN b [WHEN c THEN d]* [ELSE e] END
-- 复杂条件用 case when 更直观
select sal, case when sal<=1500 then 1
when sal<=3000 then 2
else 3 end sallevel
from emp;
-- 以下语句等价
select ename, deptno,
case deptno when 10 then 'accounting'
when 20 then 'research'
when 30 then 'sales'
else 'unknown' end deptname
from emp;
select ename, deptno,
case when deptno=10 then 'accounting'
when deptno=20 then 'research'
when deptno=30 then 'sales'
else 'unknown' end deptname
from emp;
-- COALESCE(T v1, T v2, ...)。返回参数中的第一个非空值；如果所有值都为
NULL，那么返回NULL
select sal, coalesce(comm, 0) from emp;
-- isnull(a) isnotnull(a)
select * from emp where isnull(comm);
select * from emp where isnotnull(comm);
-- nvl(T value, T default_value)
select empno, ename, job, mgr, hiredate, deptno, sal +
nvl(comm,0) sumsal
from emp;
-- nullif(x, y) 相等为空，否则为a
SELECT nullif("b", "b"), nullif("b", "a");

UDTF函数

UDTF : User Defined Table-Generating Functions。用户定义表生成函数，一行输入，多行输出

-- explode，炸裂函数
-- 就是将一行中复杂的 array 或者 map 结构拆分成多行
select explode(array('A','B','C')) as col;
select explode(map('a', 8, 'b', 88, 'c', 888));

-- SELECT pageid, explode(adid_list) AS myCol... is not supported
-- SELECT explode(explode(adid_list)) AS myCol... is not
supported

lateralView: LATERAL VIEW udtf(expression) tableAlias AS
columnAlias (',' columnAlias)*
fromClause: FROM baseTable (lateralView)*

-- lateral view 的基本使用
with t1 as (
select 'OK' cola, split('www.xxx.com', '\\.') colb
)
select cola, colc
from t1
lateral view explode(colb) t2 as colc;

-- 数据
1       1,2,3
2       2,3
3       1,2
-- 创建表
create table tab1(id int, tags string) row format delimited fields terminated by '\t';
-- 加载数据
load data local inpath '/data/hadoop/data/uid1.dat' into table tab1;
-- 查询
select id, tag from tab1 lateral view explode(split(tags, ',')) t1 as tag;

窗口函数

over 关键字
-- 查询emp表工资总和
select sum(sal) from emp;
-- 不使用窗口函数，有语法错误
select ename, sal, sum(sal) salsum from emp;
-- 使用窗口函数，查询员工姓名、薪水、薪水总和
select ename, sal, sum(sal) over() salsum,
concat(round(sal / sum(sal) over()*100, 1) || '%')
ratiosal
from emp;
partition by子句
-- 查询员工姓名、薪水、部门薪水总和
select ename, sal, sum(sal) over(partition by deptno) salsum
from emp;
order by 子句
-- 增加了order by子句；sum：从分组的第一行到当前行求和
select ename, sal, deptno, sum(sal) over(partition by deptno
order by sal) salsum
from emp;
Window子句
rows between ... and ...

#### 排名函数
#### 序列函数

### Hive 事务
> 事务(transaction)是一组单元化操作，这些操作要么都执行，要么都不执行，是一个不可分割的工作单元。
#### 事务4要素
> 原子性（Atomicity）、一致性（Consistency）、隔离性
（Isolation）、持久性（Durability），这四个基本要素通常称为ACID特性
* 原子性
  * 一个事务是一个不可再分割的工作单位，事务中的所有操作要么都发生，要么都不发生
* 一致性
  * 事务的一致性是指事务的执行不能破坏数据库数据的完整性和一致性，一个事务在执行之前和执行之后，数据库都必须处于一致性状态
* 隔离性
  * 在并发环境中，并发的事务是相互隔离的，一个事务的执行不能被其他事务干扰。即不同的事务并发操纵相同的数据时，每个事务都有各自完整的数据空间，即一个事务内部的操作及使用的数据对其他并发事务是隔离的，并发执行的各个事务之间不能互相干扰
* 持久性
  * 事务一旦提交，它对数据库中数据的改变就应该是永久性的
* 默认下，`hive`不支持事务和行级更新，需要配置
#### Hive事务的限制
* `Hive`提供行级别的`ACID`语义
* `BEGIN`、`COMMIT`、`ROLLBACK`暂时不支持，所有操作自动提交
目前只支持`ORC`的文件格式
* 默认事务是关闭的，需要设置开启
* 要是使用事务特性，表必须是分桶的
* 只能使用内部表
* 如果一个表用于`ACID`写入（`INSERT、UPDATE、DELETE`），必须在表中设置表属性 : `"transactional=true"`
* 必须使用事务管理器`org.apache.hadoop.hive.ql.lockmgr.DbTxnManager`
* 目前支持快照级别的隔离。就是当一次数据查询时，会提供一个数据一致性的快照
* `LOAD DATA`语句目前在事务表中暂时不支持

####  如何让HDFS上支持数据的更新
* HDFS是不支持文件的修改
* 并且当有数据追加到文件，HDFS不对读数据的用户提供一致性的
* 为了在HDFS上支持数据的更新
  * 表和分区的数据都被存在基本文件中（`base files`）
  * 新的记录和更新，删除都存在增量文件中（`delta files`）
  * 一个事务操作创建一系列的增量文件
  * 在读取的时候，将基础文件和修改，删除合并，最后返回给查询
#### Hive 事务操作示例

### 元数据管理与存储
#### Metastore
##### 映射
* 映射指的是一种对应关系
* 在Hive中需要描述清楚表跟文件之间的映射关系、列和字段之间的关系等等信息
* 这些描述映射关系的数据的称之为`Hive的元数据`
* 该数据十分重要，因为只有通过查询它才可以确定用户编写sql和最终操作文件之间的关系。
##### 元数据
* `Metadata`即元数据
  * 元数据包含用Hive创建的database、table、表的字段等元信息
  * 元数据存储在关系型数据库中,如hive内置的Derby、第三方如MySQL等
* `Metastore`即元数据服务
  * 是Hive用来管理库表元数据的一个服务。
  * 有了元数据服务，上层的服务不用再跟裸的文件数据打交道，而是可以基于结构化的库表信息构建计算框架
  * `metastore`服务实际上就是一种`thrift`服务，通过它用户可以获取到`Hive`元数据，并且通过`thrift`获取元数据的方式，屏蔽了数据库访问需要驱动，`url`，用户名，密码等细节
#### Metastore三种配置方式
##### 内嵌模式
* 内嵌模式使用的是内嵌的`Derby`数据库来存储元数据，也不需要额外起`Metastore`服务
* 数据库和`Metastore`服务都嵌入在主`Hive Server`进程中
* 默认的，配置简单，解压hive安装包 bin/hive 启动即可使用；
* 一次只能一个客户端连接，适用于用来实验，不适用于生产环境
* 不同路径启动`hive`，每一个`hive`拥有一套自己的元数据，无法共享。
##### 本地模式
* 本地模式采用外部数据库来存储元数据
* 目前支持的数据库有：`MySQL`、`Postgres`、`Oracle`、`MS SQL Server`
* 本地模式不需要单独起`metastore`服务，用的是跟`Hive`在同一个进程里的`metastore`服务
* `Hive`根据 `hive.metastore.uris` 参数值来判断，如果为空，则为本地模式。
* 每启动一次`hive`服务，都内置启动了一个`Metastore`
* 在`hive-site.xml`中暴露的数据库的连接信息
* 配置较简单，本地模式下`hive`的配置中指定`mysql`的相关信息即可
##### 远程模式
* 远程模式下，需要单独起`Metastore`服务
* 每个客户端都在配置文件里配置连接到该`Metastore`服务。
* 远程模式的`Metastore`服务和`hive`运行在不同的进程里。
* 在生产环境中，建议用远程模式来配置`Hive Metastore`
* 远程模式下，其他依赖hive的软件都可以通过Metastore访问Hive
* 需要配置`hive.metastore.uris`参数来指定`Metastore`服务运行的机器`ip`和端口，并且需要单独手动启动metastore服务
* `Metastore`服务可以配置多个节点上，避免单节点故障导致整个集群的`hive client`不可用
* 同时`hive client`配置多个`Metastore`地址，会自动选择可用节点。
#### 配置

### HiveServer2
> HiveServer2是一个服务端接口，使远程客户端可以执行对Hive的查询并返回结果。
* 目前基于Thrift RPC的实现是HiveServer的改进版本，并支持多客户端并发和身份验证
* 启动hiveServer2服务后，就可以使用`jdbc`、`odbc`、`thrift`的方式连接
* `Thrift`是一种接口描述语言和二进制通讯协议，它被用来定义和创建跨语言的服务。
* 被当作一个远程过程调用（`RPC`）框架来使用
* `HiveServer2（HS2）`是一种允许客户端对`Hive`执行查询的服务
#### HiveServer2作用
* 为`Hive`提供了一种允许客户端远程访问的服务
* 基于`thrift`协议，支持跨平台，跨编程语言对`Hive`访问
* 允许远程访问`Hive`
#### HiveServer2配置

### HCatalog
> `HCatalog`提供了一个统一的元数据服务，允许不同的工具如`Pig`、`MapReduce`等
* 通过`HCatalog`直接访问存储在`HDFS`上的底层文件
* `HCatalog`是用来访问`Metastore`的`Hive`子项目
* `HCatalog`使用了`Hive`的元数据存储
  * 使得像`MapReduce`这样的第三方应用可以直接从`Hive`的数据仓库中读写数据
* `HCatalog`支持用户在`MapReduce`程序中只读取需要的表分区和字段，不需要读取整个表
  * 提供一种逻辑上的视图来读取数据，而不仅仅是从物理文件的维度
* `HCatalog`提供了一个称为`hcat`的命令行工具
  * `hcat`只接受不会产生`MapReduce`任务的命令
### 数据存储格式
* Hive支持的存储数的格式主要有：
  * `TEXTFILE`（默认格式）
  * `SEQUENCEFILE`
  * `RCFILE`
  * `ORCFILE`
  * `PARQUET`
    * `TEXTFILE`为默认格式，建表时没有指定文件格式，则使用`TEXTFILE`，导入数据时，会直接把数据文件拷贝到`hdfs`上不进行处理
    * `sequencefile`，`rcfile`，`orcfile`格式的表不能直接从本地文件导入数据，数据要先导入到`textfile`格式的表中， 然后再从表中用`insert`导入`sequencefile`、`rcfile`、`orcfile`表中
#### 行存储与列存储
* 行式存储下一张表的数据都是放在一起的，但列式存储下数据被分开保存了
##### 行式存储：
* 优点：数据被保存在一起，`insert`和`update`更加容易
* 缺点：选择（`selection`）时即使只涉及某几列，所有数据也都会被读取
* `TEXTFILE`、`SEQUENCEFILE`的存储格式是基于行存储的；
##### 列式存储：
* 优点：查询时只有涉及到的列会被读取，效率高
* 缺点：选中的列要重新组装，`insert/update`比较麻烦
* `ORC`和`PARQUET`是基于列式存储
#### TextFile
* `Hive`默认的数据存储格式
* 数据不做压缩，磁盘开销大，数据解析开销大
* 可结合`Gzip`、`Bzip2`使用(系统自动检查，执行查询时自动解压)，但使用这种方式，`hive`不会对数据进行切分，从而无法对数据进行并行操作
#### SEQUENCEFILE
* `SequenceFile`是`Hadoop API`提供的一种二进制文件格式
* 其具有使用方便、可分割、可压缩的特点。
* `SequenceFile`支持三种压缩选择：`none`，`record`，`block`
* `Record`压缩率低，一般建议使用`BLOCK`压缩
#### RCFile
* `RCFile`全称`Record Columnar File`
* 列式记录文件，是一种类似于`SequenceFile`的键值对数据文件
* `RCFile`结合列存储和行存储的优缺点，是基于行列混合存储的
`RCFile`
* `RCFile`遵循的【先水平划分，再垂直划分】的设计理念
  * 先将数据按行水平划分为行组
  * 这样一行的数据就可以保证存储在同一个集群节点
  * 然后在对行进行垂直划分

## 数据采集工具Flume
上一篇下一篇

猜你喜欢

热点阅读