大数据之hbase_Phoenix教程_hbase phoenix教程

1简介

Apache Phoneix是运行在HBase之上的高性能关系型数据库，通过Phoneix可以像使用jdbc访问关系型数据库一样访问HBase。Phoneix操作的表以及数据存储在HBase上，phoneix只需要和HBase进行表关联。然后在用工具进行一些读写操作。可以把Phoneix只看成一种代替HBase语法的工具(put get scan 过滤器 ,delete)。可以使用java的jdbc来连接phoneix操作Hbase数据库。phoenix在查询hbase时，虽然做了一些优化，但是延迟还是不小。所以依然用在OLAT中，在将结果返回存储下来。Apache Phoenix 与其他 Hadoop 产品完全集成，如 Spark，Hive，Pig，Flume 和 MapReduce。
hbase中存储的是Hfile文件
当今的数据处理大致可以分成两大类：联机事务处理OLTP（on-line transaction processing）、联机分析处理OLAP（On-Line Analytical Processing）。OLTP是传统的关系型数据库的主要应用，主要是基本的、日常的事务处理，例如银行交易。OLAP是数据仓库系统的主要应用，支持复杂的分析操作，侧重决策支持，并且提供直观易懂的查询结果.!
1.1特点
通过JDBC API实现了大部分的java.sql接口，包括元数据API
DDL支持：通过CREATE TABLE、DROP TABLE及ALTER TABLE来添加/删除
DML支持：用于逐行插入的UPSERT VALUES,用于相同或不同表之间大量数据传输的UPSERT SELECT,用于删除行的DELETE
事务支持：通过客户端的批处理实现的有限的事务支持(beta测试中)
二级索引支持：为了多维度查询速度
遵循ANSI SQL标准

2安装

下载地址 http://phoenix.apache.org/download.html
上传 解压

复制jar包
将phoenix-5.0.0-HBase-2.0-server.jar复制到所有的hbase节点的lib目录下

cp  phoenix-5.0.0-HBase-2.0-server.jar  /usr/apps/hbase-2.0.4/lib/
scp  phoenix-5.0.0-HBase-2.0-server.jar  doit02:/usr/apps/hbase-2.0.4/lib/
scp  phoenix-5.0.0-HBase-2.0-server.jar  doit03:/usr/apps/hbase-2.0.4/lib/
1
2
3

修改hbase的配置文件

<configuration>
<!-- 指定hbase在HDFS上存储的路径 -->
<property>
<name>hbase.rootdir</name>
<value>hdfs://doit01:9000/hbase</value>
</property>
<!-- 指定hbase是分布式的 -->
<property>
<name>hbase.cluster.distributed</name>
<value>true</value>
</property>
<!-- 指定zk的地址，多个用“,”分割 -->
<property>
<name>hbase.zookeeper.quorum</name>
<value>doit01:2181,doit02:2181,doit03:2181</value>
</property>
<!--使用Phoenix建立hbase二级索引配置 -->
<property>  
    <name>hbase.regionserver.wal.codec</name>  
    <value>org.apache.hadoop.hbase.regionserver.wal.IndexedWALEditCodec</value>
</property>
<property>  
    <name>hbase.region.server.rpc.scheduler.factory.class</name>  
    <value>org.apache.hadoop.hbase.ipc.PhoenixRpcSchedulerFactory</value>  
    <description>Factory to create the Phoenix RPC Scheduler that uses separate queues for index and metadata updates</description>
</property>
<property> 
    <name>hbase.rpc.controllerfactory.class</name>  
    <value>org.apache.hadoop.hbase.ipc.controller.ServerRpcControllerFactory</value>  
    <description>Factory to create the Phoenix RPC Scheduler that uses separate queues for index and metadata updates</description>
</property>
<property>  
    <name>hbase.master.loadbalancer.class</name>  
    <value>org.apache.phoenix.hbase.index.balancer.IndexLoadBalancer</value>
</property>
<property>  
    <name>hbase.coprocessor.master.classes</name>  
    <value>org.apache.phoenix.hbase.index.master.IndexMasterObserver</value>
</property>
<!--使用 schema 和 Namespace 对应的配置 -->
<property>
<name>phoenix.schema.isNamespaceMappingEnabled</name>
<value>true</value>
</property>
<property>
<name>phoenix.schema.mapSystemTablesToNamespace</name>
<value>true</value>
</property>

</configuration>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

将hbase/conf/hbase-site.xml 复制到phoenix中的bin/目录下
重启hbase(时间 , HDFS)
stop-hbase
删除ZK的/hbase
删除hdfs的/hbase目录
start-hbase
启动Phoenix
连接到Hbase [root@doit01 bin]# ./sqlline.py doit01,doit02,doit03:2181

3操作

查看table列表  !tables  !table
查看表字段信息  !describe tablename
查看执行历史  !history
查看table 的索引  !index tablename
1
2
3
4

其他操作 help
3.1基本操作
3.1.1建表

create table demo1(
rk varchar primary key ,
name varchar ,
age integer 
) ;
1
2
3
4
5

CREATE TABLE  do_movie (
movie char(6) not null ,
rate double ,
timeStimep varchar ,
uud  VARCHAR
CONSTRAINT my_rk PRIMARY KEY (movie, timeStimep)  多字段为主键
);
1
2
3
4
5
6
7

3.1.2插入/更新数据
如果不存在则插入，否则更新表中的值。列的列表是可选的，如果不存在，则值将按照在架构中声明的顺序映射到该列。值必须计算为常数。
ON DUPLICATE KEY如果您需要UPSERT成为原子的，请使用子句（在Phoenix 4.9中可用）。在这种情况下，由于提交完成后需要在服务器端读取该行，因此性能会降低。使用IGNORE，如果你不想要的UPSERT，如果行已存在执行。否则，使用UPDATE，将对表达式求值并将结果用于设置列，例如执行原子增量。一个UPSERT以相同的同一行提交的批处理将在执行顺序进行处理。

UPSERT INTO TEST VALUES('foo','bar',3);
UPSERT INTO TEST(NAME,ID) VALUES('foo',123);
UPSERT INTO TEST(ID, COUNTER) VALUES(123, 0) ON DUPLICATE KEY UPDATE COUNTER = COUNTER + 1;
UPSERT INTO TEST(ID, MY_COL) VALUES(123, 0) ON DUPLICATE KEY IGNORE;
1
2
3
4

3.1.3查询数据

从一个或多个表中选择数据。UNION ALL合并来自多个select语句的行。ORDER BY根据给定的表达式对结果进行排序。LIMIT（或FETCH FIRST）限制查询返回的行数，如果未指定或指定为null或小于零，则不应用限制。的LIMIT（或FETCH FIRST）子句后执行ORDER BY子句支撑顶部-N型的查询。OFFSET子句在开始返回行之前会跳过很多行。可选提示可用于覆盖查询优化器做出的决策。
例：

SELECT * FROM TEST LIMIT 1000;
SELECT * FROM TEST LIMIT 1000 OFFSET 100;
SELECT full_name FROM SALES_PERSON WHERE ranking >= 5.0
  UNION ALL SELECT reviewer_name FROM CUSTOMER_REVIEW WHERE score >= 8.0
1 where 
2 order by desc 
3 limit
4 group by 
5 支持聚合函数  sum  avg  count  max  min
6 支持多表查询
1
2
3
4
5
6
7
8
9
10

3.1.4修改表结构
添加一个列

alter table demo1  add gender varchar ;
1

是删除一列

alter table demo1 drop column gender ;
1

3.1.5查看表结构

!describe  demo1 ;
1

+------------+--------------+-------------+--------------+------------+------------+--------------+----------------+--+
| TABLE_CAT  | TABLE_SCHEM  | TABLE_NAME  | COLUMN_NAME  | DATA_TYPE  | TYPE_NAME  | COLUMN_SIZE  | BUFFER_LENGTH  |  |
+------------+--------------+-------------+--------------+------------+------------+--------------+----------------+--+
|            |              | DEMO1       | RK           | 12         | VARCHAR    | null         | null           |  |
|            |              | DEMO1       | NAME         | 12         | VARCHAR    | null         | null           |  |
|            |              | DEMO1       | AGE          | 4          | INTEGER    | null         | null           |  |
|            |              | DEMO1       | GENDER       | 12         | VARCHAR    | null         | null           |  |
+------------+--------------+-------------+--------------+------------+------------+--------------+----------------+--+
1
2
3
4
5
6
7
8

3.1.6删除数据

3.2高级操作
3.2.1使用多列族建表
建议标志符的地方使用双引号 确定大小写

create table  demo2(
rk varchar primary key ,
f1.name varchar ,
f1.age integer ,
f2.hobby varchar 
) ;
1
2
3
4
5
6

查询表信息

select table_name , column_family , column_name from system.catalog where table_name='DEMO2' ;
1

+-------------+----------------+--------------+
| TABLE_NAME  | COLUMN_FAMILY  | COLUMN_NAME  |
+-------------+----------------+--------------+
| DEMO2       |                |              |
| DEMO2       | 0              |              |
| DEMO2       | F1             | AGE          |
| DEMO2       | F2             | HOBBY        |
| DEMO2       | F1             | NAME         |
| DEMO2       |                | RK           |
+-------------+----------------+--------------+
1
2
3
4
5
6
7
8
9
10

3.2.2使用namespace建表
需要在hbase中的配置文件中配置Phoenix的schema信息 , 同时要将hbase-site.xml文件复制到Phoenix的bin目录下 ;
1)在hbase中创建namespace
2)在Phoenix客户端创建对应的schema
3)切换schema
4)创建表
5)在hbase中查看

在hbase的客户端创建一个namespace

在Phoenix的客户端创建schema

create schema  "doit" 
1

切换

use "doit" ;
1

建表

create table demo3(id integer primary key , name varchar) ;
1

查看 表创建在了 doit名称空间下

3.2.3预分region建表
在学习Hbase阶段对hbase预分region的意义做了详细介绍

create table TEST2 (host varchar not null primary key, description varchar) split on ('cs','eu','na');

list_regions  "doit:TEST2"
1
2
3

3.2.4二级索引
大量的数据信息在表中在我们对数据进行查询的时候是全局的全表检索 , 影响执行效率 !
0: jdbc:phoenix:doit01,doit02,doit03:2181> explain select * from demo1 ;
±----------------------------------------------------------------±----------------±---------------±-------------+
| PLAN | EST_BYTES_READ | EST_ROWS_READ | EST_INFO_TS |
±----------------------------------------------------------------±----------------±---------------±-------------+
| CLIENT 1-CHUNK PARALLEL 1-WAY ROUND ROBIN FULL SCAN OVER DEMO1 | null | null | null |
±----------------------------------------------------------------±----------------±---------------±-------------+

创建二级索引表 , 以后再根据制定字段查询的时候就会从索引中直接获取数据, 而不是全表检索数据!
在表demo1上的NAME字段上建立索引

 create index index2 on demo1(NAME)  include (AGE);
1

查看执行计划

0: jdbc:phoenix:doit01,doit02,doit03:2181> explain select * from demo1 where NAME='zss' ;
1

+---------------------------------------------------------------------------+-----------------+----------------+------+
|                                   PLAN                                    | EST_BYTES_READ  | EST_ROWS_READ  | EST_ |
+---------------------------------------------------------------------------+-----------------+----------------+------+
| CLIENT 1-CHUNK PARALLEL 1-WAY ROUND ROBIN RANGE SCAN OVER INDEX2 ['zss']  | null            | null           | null |
+---------------------------------------------------------------------------+-----------------+----------------+------+
1 row selected (0.072 seconds)
1
2
3
4
5
6

0: jdbc:phoenix:doit01,doit02,doit03:2181> explain select * from demo1 where age=21 ;
1

+------------------------------------------------------------------+-----------------+----------------+--------------+
|                               PLAN                               | EST_BYTES_READ  | EST_ROWS_READ  | EST_INFO_TS  |
+------------------------------------------------------------------+-----------------+----------------+--------------+
| CLIENT 1-CHUNK PARALLEL 1-WAY ROUND ROBIN FULL SCAN OVER INDEX2  | null            | null           | null         |
|     SERVER FILTER BY "AGE" = 21                                  | null            | null           | null         |
+------------------------------------------------------------------+-----------------+----------------+--------------+
1
2
3
4
5
6

3.3hbase表和Phoenix映射
将hbase中的表映射到Phoenix中
3.3.1视图映射(只读)

create  view “user”(
id varchar primary key ,
"F"."NAME" varchar ,
"F"."AGE" varchar
)  ;
1
2
3
4
5

不能对数据修改 , 删除视图以后hbase中的数据不受影响

3.3.2表映射(读写)
强关联 ,当删除表的时候hbase中的数据会被删除

create  table demo5 (
id varchar primary key ,
"F"."NAME" varchar ,
"F"."AGE" varchar
) 
1
2
3
4
5

COLUMN_ENCODED_BYTES = 0 ; – 禁止列的强制映射
3.4JDBC
添加Phoenix的pom依赖或者jar包

<dependency>
    <groupId>org.apache.phoenix</groupId>
    <artifactId>phoenix-core</artifactId>
    <version>5.0.0-HBase-2.0</version>
</dependency>
1
2
3
4
5

public static void main(String[] args) throws SQLException {
//创建属性对象
    Properties p = new Properties();
// 添加namespace的支持
    p.setProperty("phoenix.schema.isNamespaceMappingEnabled", "true");
    Connection conn = DriverManager.getConnection("jdbc:phoenix:doit01:2181", p);
    Statement st = conn.createStatement();
  /*  int i = st.executeUpdate("create  schema  dodo");
    conn.commit();*/

    ResultSet resultSet = st.executeQuery("select * from \"t_y\"");
    while(resultSet.next()){
        String string = resultSet.getString(1);
        String string2 = resultSet.getString(1);
        System.out.printf(string+string2);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

4phoenix中的函数

http://phoenix.apache.org/language/functions.html
1

5H综合案例

{“movie”:“1193”,“rate”:“5”,“timeStamp”:“978300760”,“uid”:“1”}
{“movie”:“661”,“rate”:“3”,“timeStamp”:“978302109”,“uid”:“1”}
将100万条数据导入到hbase表中 ,查询数据 ! 比较rowkey查询和普通字段查询时间差值!
怎么存储?关键是设计rowkey(考虑到插入的和查询的热点 , 查询效率 查询的维度) ;
电影_时间(rowkey)[定长] 0001 0002 0012
用户_时间(rowkey)[定长]
实现方式:
1 MR(java)程序读取行数据 封装成put对象存储到hbase表中
2 MR程序读取数据 将数据转换成Hfile数据 , 再到入到hbase表中(shell )
3读取数据直接写到hbase表中
4 java程序读取数据 ,预处理成csv文件格式 , shell创建hfile , 再导入数据

使用Phoenix创建映射表关联hbase表数据使用sql查询 , 建立索引表, 查询索引字段和非索引西段差值

5.1本地读取movie电影数据输出csv文件

/**
 * 读取本地的电影评分数据<br/>
 *  将数据转换哼csv文件
 * 
 * @author ThinkPad
 *
 */
public class LoadMovieData {
	@SuppressWarnings("resource")
	public static void main(String[] args) throws IOException {
		Gson gs = new Gson();
		// 读取movie数据
		BufferedReader br = new BufferedReader(new FileReader("D:\\data\\movie\\input\\rating.json"));
		// 输出流
		BufferedWriter bw = new  BufferedWriter(new  FileWriter("d://movie.csv")) ;
		String line = null;
		while ((line = br.readLine()) != null) {
			try {
				// 将每行数据转换成moviebean
				MovieBean mb = gs.fromJson(line, MovieBean.class);
				// 获取属性信息
				String movie = mb.getMovie();
				double rate = mb.getRate();
				String timeStamp = mb.getTimeStamp();
				String uid = mb.getUid();
				String row_key = StringUtils.leftPad(movie, 5, "0")+"_"+timeStamp ;
				String new_uid = StringUtils.leftPad(uid, 4, "0")+"_"+timeStamp ;
				String key = row_key+","+movie+","+rate+","+timeStamp+","+new_uid ;
				bw.write(key);
				bw.newLine();
			} catch (Exception e) {
				continue ;
			}
		}
		bw.close();
		br.close();
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

5.2将csv文件导入到hbase表中
在Hbase中创建movie 表

create “movie” , “F”
1

将生成的数据上传到HDFS中
使用生成Hfile文件

hbase  org.apache.hadoop.hbase.mapreduce.ImportTsv \
-Dimporttsv.separator=, \
-Dimporttsv.columns='HBASE_ROW_KEY,F:MOVIE,F:RATE,F:TIMESTAMP,F:UID'  \
-Dimporttsv.bulk.output=/data/output \
movie  /data/movie.csv
1
2
3
4
5

将数据导入到hbase表中

hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles /data/output movie
1

5.3查询数据查看时间
5.4创建Phoenix表映射hbase表
create view “movie”(
rk varchar primary key ,
“cf”.“movie” varchar ,
“cf”.“rate” varchar ,
“cf”.“timeStamp” varchar ,
“cf”.“uid” varchar
) ;
主键查询

select *  from "movie" where substr(rk,0,5) = '00003' ;
select *  from "movie" where substr(rk,0,5) = '00003' limit 1;
1 row selected (0.074 seconds)
0: jdbc:phoenix:doit01:2181> select count(*)  from "movie" where substr(rk,0,5) = '00003' ;
0: jdbc:phoenix:doit01:2181> select avg(to_number("rate"))  from "movie" where substr(rk,0,5) = '00003' ;
1
2
3
4
5

非主键查询

select * from "movie" where "uid"='2909' limit 1;
1 row selected (0.156 seconds)
1
2

映射的数据量比较大,可能出现请求超时, 将请求超时时间配置大一点
在Phoenix中bin的hbase-site.xml中配置

 <property>
        <name>phoenix.query.timeoutMs</name>
        <value>1800000</value>
    </property>    
    <property>
        <name>hbase.regionserver.lease.period</name>
        <value>1200000</value>
    </property>
    <property>
        <name>hbase.rpc.timeout</name>
        <value>1200000</value>
    </property>
    <property>
        <name>hbase.client.scanner.caching</name>
        <value>1000</value>
    </property>
    <property>
        <name>hbase.client.scanner.timeout.period</name>
        <value>1200000</value>
    </property>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

5.5创建索引
1 直接建立一个索引表 不包含任何的查询索引字段

create index "movie_index2" on "movie"("cf"."uid") ;
select uid from “movie” where “uid”=’1’
1
2

2 直接建立一个索引表 包含任何的查询索引字段
注意: 以后根据uid查询的时候如果查询的字段在include中,查询走索引

create index "movie_index3" on "movie"("cf"."uid") include("cf"."rate") ;
1

select uid , rate  from “movie” where “uid”=’1’  走索引 块
1

全表检索

select * from “movie” where “uid”=’1’ 
select timeStamp from “movie” where “uid”=’1’ 
1
2

查看执行计划

explain  select "timeStamp"  from "movie" where "uid" = '1' limit 2 ;
1

全表扫面

走索引

rowkey