Hadoop---(5)Hive（数据仓库）_hive数仓搭建二手房信息

 

5.1 Hive的系统架构

 

用户接口，包括CLI，JDBC/ODBC，WebUI。

元数据存储，通常是存储在关系数据库如mysql，derby中。

解释器、编译器、优化器、执行器。

Hadoop：用HDFS进行存储，利用MapReduce进行计算。

 

 

 

 

1.     用户接口主要有三个：CLI，JDBC/ODBC和WebUI

1. CLI，即Shell命令行。
2. JDBC/ODBC是hive的java，与使用传统数据JDBC的方式类似。
3. WebUI是通过浏览器访问Hive

 

2.     Hive将元数据存储在数据库中（metastore），目前只支持mysql，derby。

3.     Hive中元数据包括表的名字，表的列和分区以及属性，表的属性（是否为外部表等），表的数据所在目录等…

4.     解释器、编译器、优化器完成HQL查询语句从词法分析、语法分析、编译、优化以及查询计划（plan）的生成。生成的查询计划存储在HDFS中，并在随后有MapReduce调用执行。

5.     Hive的数据存储在HDFS中，大部分的查询由MapReduce完成（包含*的查询，比如select * from table不会生成MapReduce任务）

 

 

Compiler：

1. Driver调用编译器（compiler）处理HiveQL字符串，这些字符串可能是一条DDL、DML或查询语句。
2. 编译器将字符串转换为策略（plan）
3. 策略仅有元数据操作和HDFS操作完成，元数据操作只包含DDL语句，HDFS操作只包含LOAD语句。
4. 对插入和查询而言，策略由map-reduce任务中具有方向的非循环图（directedacyclic graph , DAG）组成。

 

 

 

5.2 Hive的metastore

 

metastore是hive元数据的集中存放地。metastore默认使用内嵌的derby数据库作为存储引擎。

Derby引擎的缺点：一次只能打开一个会话。

 

使用Mysql作为外置存储引擎，多用户同时访问。

 

 

 

5.3 Hive的运行模式

 

Hive的运行模式即任务的执行环境。

分为本地与集群两种。

我们可以通过Mapred.job.tracker来指明

设置方式：

Hive> SET mapred.job.tracker=local

 

 

 

 

 

5.4 Hive的启动方式

 

1. hive 命令行模式，直接输入#/hive/bin/hive的执行程序，或者输入 #hive --service cli
2. hive web界面的 (端口号9999) 启动方式

#hive --service hwi &

用于通过浏览器来访问hive

http://hadoop0:9999/hwi/

1. hive 远程服务 (端口号10000) 启动方式

#hive --service hiveserver &

 

 

5.5 Hive与传统数据库

 

 

 

Hive 中没有定义专门的数据格式，数据格式可以由用户指定，用户定义数据格式需要指定三个属性：列分隔符（通常为空格、”\t”、”\x001″）、行

分隔符 （”\n”）以及读取文件数据的方法（Hive 中默认有三个文件格式 TextFile，SequenceFile 以及 RCFile）。

由于在加载数据的过程中，不需要从用户数据格式到 Hive 定义的数据格式的转换，因此，Hive 在加载的过程中不会对数据本身进行任何修改，而只是将数据内容复制或者移动到相应的 HDFS 目录中。

而在数据库中，不同的数据库有不同的存储引擎，定义了自己的数据格式。所有数据都会按照一定的组织存储，因此，数据库加载数据的过程会比较耗时。

 

 

 

5.6 Hive的数据类型

 

 

基本数据类型

Tinyint/smallint/int/bigint

Float/double

Boolean

String

复杂数据类型

Array/Map/Struct

没有date/dateTime

 

 

5.7 Hive的数据存储

 

Hive的数据存储基于Hadoop HDFS

Hive没有专用的数据存储格式

存储结构主要包括：数据库、文件、表、视图

Hive默认可以直接加载文本文件（Text File），还支持sequence file

创建表时，指定hive数据的列分隔符与行分隔符，Hive即可解析数据

 

 

 

5.8 Hive的数据模型

 

5.8.1 Hive的数据模型-数据库

 

类似传统数据库的DataBase

默认数据库“default”

使用#hive命令后，不使用hive>use<数据库名>，系统默认的数据库。

可以显式使用hive> user default;

 

创建一个新数据库

Hive>create database  test_dw;

 

 

5.8.2 Hive的数据模型-表

 

Table内部表

Partition分区表

External Table外部表

Bucket Table桶表

 

 

5.8.3 Hive的数据模型-内部表

 

1. 与数据库中的Table在概念上是类似。
2. 每一个Table在Hive中都有一个相应的目录存储数据。

例如：一个表test，它在HDFS中的路径为：/warehouse/test。Warehouse是在hive-site.xml中由${hive.metastore.warehouse.dir}指定的数据仓库的目录。

1. 所有的Table数据（不包括External Table）都保存在这个目录中。
2. 删除表时，元数据和数据都会被删除。

 

 

创建数据文件inner_table.dat

 

1. 创建表

Hive>create table inner_table(key string);

 

1. 加载数据

Hive>load data local inpath ‘/root/inner_table.dat’ into table inner_table;

 

1. 查看数据

select * from inner_table;

select count(*) from inner_table;

 

1. 删除表

drop table inner_table

 

删除表时可能报错，max key length is 1000 bytes

把mysql的数据库字符串类型改为latin1

 

 

5.8.4 Hive的数据模型-分区表

 

Partition对应数据库的partition列的密集索引。

在hive中，表中的一个partition对应于表下的一个目录，所有的partition的数据都存储在对应的目录中。

 

例如：test表中包含 date 和 city 两个 Partition，

则对应于date=20130201, city = bj 的 HDFS 子目录为：

/warehouse/test/date=20130201/city=bj

对应于date=20130202, city=sh 的HDFS 子目录为；

/warehouse/test/date=20130202/city=sh

 

 

示例：

CREATE TABLE tmp_table #表名

(

title   string, # 字段名称 字段类型

minimum_bid     double,

quantity        bigint,

have_invoice    bigint

)COMMENT '注释：XXX' #表注释

 PARTITIONED BY(pt STRING) #分区表字段（如果你文件非常之大的话，采用分区表可以快过滤出按分区字段划分的数据）

 ROW FORMAT DELIMITED

   FIELDS TERMINATED BY '\001'   # 字段是用什么分割开的

STORED AS SEQUENCEFILE; #用哪种方式存储数据，SEQUENCEFILE是hadoop自带的文件压缩格式

 

 

一些相关命令

SHOW TABLES; # 查看所有的表

SHOW TABLES '*TMP*'; #支持模糊查询

SHOW PARTITIONS TMP_TABLE; #查看表有哪些分区

DESCRIBE TMP_TABLE; #查看表结构

 

 

1. 创建数据文件partition_table.dat
2. 创建表

create table partition_table(rectime string,msisdn string) partitioned by(daytime string,city string) row format delimited fields terminated by '\t' stored as TEXTFILE;

1. 加载数据到分区

load data local inpath '/home/partition_table.dat' into table partition_table partition (daytime='2013-02-01',city='bj');

1. 查看数据

select * from partition_table

select count(*) from partition_table

•删除表 drop table partition_table

 

 

1. alter table partition_table add partition (daytime='2013-02-04',city='bj');

通过load data 加载数据

1. alter table partition_table drop partition (daytime='2013-02-04',city='bj')

元数据，数据文件删除，但目录daytime=2013-02-04还在。

 

 

5.8.5 Hive的数据模型-桶表

 

1. 桶表是对数据进行哈希取值，然后放到不同文件中存储。
2. 创建表

create table bucket_table(id string) clustered by(id) into 4 buckets;            

1. 加载数据

set hive.enforce.bucketing = true;

insert into table bucket_table select name from stu;  

insert overwrite table bucket_table select name from stu;

1. 数据加载到桶表时，会对字段取hash值，然后与桶的数量取模。把数据放到对应的文件中。
2. 抽样查询

select * from bucket_table tablesample(bucket 1 out of 4 on id);

5.8.6 Hive的数据模型-外部表

 

1. 指向已经在 HDFS 中存在的数据，可以创建 Partition
2. 它和 内部表 在元数据的组织上是相同的，而实际数据的存储则有较大的差异
3. 内部表 的创建过程和数据加载过程（这两个过程可以在同一个语句中完成），在加载数据的过程中，实际数据会被移动到数据仓库目录中；之后对数据对访问将会直接在数据仓库目录中完成。删除表时，表中的数据和元数据将会被同时删除
4. 外部表 只有一个过程，加载数据和创建表同时完成，并不会移动到数据仓库目录中，只是与外部数据建立一个链接。当删除一个 外部表 时，仅删除该链接

CREATE EXTERNAL TABLE page_view

( viewTime INT,

  userid BIGINT,

  page_url STRING,

 referrer_url STRING,                                               

  ip STRING COMMENT 'IP Address of the User',

  country STRING COMMENT 'country of origination‘

)

    COMMENT 'This is the staging page view table'

    ROW FORMAT DELIMITED FIELDS TERMINATED BY '44' LINES    TERMINATED BY '12'

    STORED AS TEXTFILE

    LOCATION 'hdfs://centos:9000/user/data/staging/page_view';

 

1. 创建数据文件external_table.dat
2. 创建表

hive>create external table external_table1 (key string) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' location '/home/external';

在HDFS创建目录/home/external

#hadoop fs -put /home/external_table.dat /home/external

1. 加载数据

LOAD DATA INPATH '/home/external_table1.dat' INTO TABLE external_table1;

1. 查看数据

select * from external_table

select count(*) from external_table

1. 删除表

drop table external_table

 

5.9 视图操作

 

5.9.1 视图的创建

 

CREATE VIEW v1 AS select * from t1;

 

5.9.2 表的修改

 

alter table target_tab add columns (cols,string)

 

5.9.3 表的删除

 

drop table

 

 

5.10 导入数据

 

当数据被加载至表中时，不会对数据进行任何转换。Load 操作只是将数据复制/移动至 Hive 表对应的位置。

 

LOAD DATA [LOCAL] INPATH 'filepath' [OVERWRITE]

INTO TABLE tablename

 [PARTITION (partcol1=val1, partcol2=val2 ...)]

 

把一个Hive表导入到另一个已建Hive表

INSERT OVERWRITE TABLE tablename [PARTITION (partcol1=val1, partcol2=val2 ...)] select_statement FROM from_statement

 

CTAS

CREATE [EXTERNAL] TABLE [IF NOT EXISTS] table_name

(col_name data_type, ...) …

AS SELECT …

 

例：create table new_external_test as  select * from external_table1;

 

 

 

5.11 查询

 

 

select

SELECT [ALL | DISTINCT] select_expr, select_expr, ...

FROM table_reference

[WHERE where_condition]

[GROUP BY col_list]

[ CLUSTER BY col_list | [DISTRIBUTE BY col_list] [SORT BY col_list] | [ORDER BY col_list] ]

[LIMIT number]

 

 

DISTRIBUTE BY 指定分发器（Partitioner）,多Reducer可用

 

 

5.11.1 基于Partition的查询 

 

一般 SELECT 查询是全表扫描。但如果是分区表，查询就可以利用分区剪枝（input pruning）的特性，类似“分区索引“”，只扫描一个表中它关心的那一部分。Hive 当前的实现是，只有分区断言（Partitioned by）出现在离 FROM 子句最近的那个WHERE 子句中，才会启用分区剪枝。例如，如果 page_views 表（按天分区）使用 date 列分区，以下语句只会读取分区为‘2008-03-01’的数据。

 SELECT page_views.*    FROM page_views    WHERE page_views.date >= '2013-03-01' AND page_views.date <= '2013-03-01'

 

5.11.2 LIMIT Clause

 

Limit 可以限制查询的记录数。查询的结果是随机选择的。下面的查询语句从 t1 表中随机查询5条记录：

SELECT * FROM t1 LIMIT 5

 

5.11.3 Top N查询

 

下面的查询语句查询销售记录最大的 5 个销售代表。

SET mapred.reduce.tasks = 1
  SELECT * FROM sales SORT BY amount DESC LIMIT 5

 

 

5.12 表连接

 

1. 导入ac信息表

hive> create table acinfo (name string,acip string)  row format delimited fields terminated by '\t' stored as TEXTFILE;

hive> load data local inpath '/home/acinfo/ac.dat' into table acinfo;

1. 内连接

select b.name,a.* from dim_ac a join acinfo b on (a.ac=b.acip) limit 10;

1. 左外连接

select b.name,a.* from dim_ac a left outer join acinfo b on a.ac=b.acip limit 10;

 

 

 

5.13 JAVA客户端

 

1. Hive远程服务启动#hive --service hiveserver >/dev/null  2>/dev/null &
2. JAVA客户端相关代码

Class.forName("org.apache.hadoop.hive.jdbc.HiveDriver");

Connection con = DriverManager.getConnection("jdbc:hive://192.168.1.102:10000/wlan_dw", "", "");

Statement stmt = con.createStatement();

String querySQL="SELECT * FROM wlan_dw.dim_m order by flux desc limit 10";

ResultSet res = stmt.executeQuery(querySQL); 

while (res.next()) {

System.out.println(res.getString(1) +"\t" +res.getLong(2)+"\t" +res.getLong(3)+"\t" +res.getLong(4)+"\t" +res.getLong(5));

}

 

 

5.14 UDF

 

1、UDF函数可以直接应用于select语句，对查询结构做格式化处理后，再输出内容。

2、编写UDF函数的时候需要注意一下几点：

a）自定义UDF需要继承org.apache.hadoop.hive.ql.UDF。

b）需要实现evaluate函数，evaluate函数支持重载。

4、步骤

a）把程序打包放到目标机器上去；

b）进入hive客户端，添加jar包：hive>add jar /run/jar/udf_test.jar;

c）创建临时函数：hive>CREATE TEMPORARY FUNCTION add_example AS 'hive.udf.Add';

d）查询HQL语句：

SELECT add_example(8, 9) FROM scores;

SELECT add_example(scores.math, scores.art) FROM scores;

SELECT add_example(6, 7, 8, 6.8) FROM scores;

e）销毁临时函数：hive> DROP TEMPORARY FUNCTION add_example;

注：UDF只能实现一进一出的操作，如果需要实现多进一出，则需要实现UDAF

 

 

 

5.15 Hive操作

 

 

 

 

创建数据库(使用SCHEMA方式)

CREATE SCHEMA userdb;

 

创建数据库

CREATE DATABASE userdb;

 

创建数据库（存在则不创建，不存在则创建）

CREATE DATABASE IF NOT EXISTS userdb;

 

列出数据库列表

SHOW DATABASES;

 

删除数据库

DROP DATABASE IF EXISTS userdb;

 

删除数据库(全部删除相应的表在删除数据库之前)

DROP DATABASE IF EXISTS userdb CASCADE;

 

删除数据库(使用SCHEMA方式)

DROP SCHEMA userdb;

 

 

 

 

 

创建表

CREATE TABLE IF NOT EXISTS employee (eid int, name String ,

salary String, destination String)

COMMENT 'Employee details'

ROW FORMAT DELIMITED

FIELDS TERMINATED BY '\t'

LINES TERMINATED BY '\n'

STORED AS TEXTFILE;

 

修改表名

ALTER TABLE employee RENAME TO emp;

 

修改列名和列数据类型

ALTER TABLE emp CHANGE name ename String;

 

增加列（列名为dept，类型为String）

ALTER TABLE emp ADD COLUMNS(

dept STRING COMMENT 'Department name');

 

删除列

ALTER TABLE employee DROP COLUMN destination;

 

使用empid代替eid列，name代替ename列

ALTER TABLE emp REPLACE COLUMNS(

eid INT empid Int,

ename STRING name String);

 

删除表

DROP TABLE IF EXISTS emp;

 

查看所有表

SHOW TABELS;

 

 

 

 

 

smple.txt

1201       Gopal    45000    Technical     manager

1202       Manisha      45000    Proof     reader

1203       Masthanvali 40000    Technical     writer

1204       Kiran     40000    Hr   Admin

1205       Kranthi  30000    Op  Admin

 

 

 

插入数据(将文本插入表中)

LOAD DATA LOCAL INPATH '/home/hadoop/smple.txt'

OVERWRITE INTO TABLE employee;

 

插入数据

insert into employee values(102,'ss','ddd','dd');

 

删除数据

 

 

查询一条数据

select * from employee where eid=1204;

 

查询所有数据

select * from employee;

 

 

 

 

 

创建视图（为工资超过30000的创建一个视图）

CREATE VIEW emp_30000 AS

SELECT * FROM employee

WHERE salary>30000;

 

查看视图

select * from emp_30000;

 

删除视图

DROP VIEW emp_30000;

 

创建索引（对salary列创建一个索引）

CREATE INDEX index_salary ON TABLE employee(salary)

AS 'org.apache.hadoop.hive.ql.index.compact.CompactIndexHandler'

with deferred rebuild

IN TABLE index_salary_employee;

 

删除索引(删除index_salary索引)

DROP INDEX index_salary ON employee;

 

升序排列

SELECT eid, name FROM employee ORDER BY name ASC;

 

降序排列

SELECT eid, name FROM employee ORDER BY name DESC;

 

分组

SELECT name, count(*) FROM employee GROUP BY name;

 

 

 

连接查询

 

JOIN（内连接）

SELECT c.id, c.name, c.age, o.amount

FROM CUSTOMERS c JOIN ORDERS o

ON (c.id = o.CUSTOMERS);

 

LEFT OUTER JOIN（左外连接）

SELECT c.id, c.name, c.age, o.amount

FROM CUSTOMERS c

LEFT OUTER JOIN ORDERS o

ON (c.id = o.CUSTOMERS);

 

RIGHT OUTER JOIN（右外连接）

SELECT c.id, c.name, c.age, o.amount

FROM CUSTOMERS c

RIGHT OUTER JOIN ORDERS o

ON (c.id = o.CUSTOMERS);

 

FULL OUTER JOIN（全连接）

SELECT c.id, c.name, c.age, o.amount

FROM CUSTOMERS c

FULL OUTER JOIN ORDERS o

ON (c.id = o.CUSTOMERS);