Phoenix on HBase

Posted 2020-10-01 累吗？累就对了，舒服是留给死人的...

（一）概要

Apache Phoenix是基于BSD许可开源的一个Java中间层，可以让开发者在Apache HBase上执行SQL查询。Apache Phoenix主要特性：

• 嵌入式的JDBC驱动，实现了大部分的java.sql接口，包括元数据API 
• 可以通过多部行键或是键/值单元对列进行建模 
• 完善的查询支持，可以使用多个谓词以及优化的扫描键 
• DDL支持：通过CREATE TABLE、DROP TABLE及ALTER TABLE来添加/删除列 
• 版本化的模式仓库：当写入数据时，快照查询会使用恰当的模式 
• DML支持：用于逐行插入的UPSERT VALUES、用于相同或不同表之间大量数据传输的UPSERT SELECT、用于删除行的DELETE 
• 通过客户端的批处理实现的有限的事务支持 
• 单表——还没有连接，同时二级索引也在开发当中 
• 紧跟ANSI SQL标准，HIVE不完全支持SQL92，而Phoenix 接近ANSI SQL-2003

（二）安装

HBase Compatibility

• Phoenix 2.x - HBase 0.94.x
• Phoenix 3.x - HBase 0.94.x
• Phoenix 4.x - HBase 0.98.1+

　　从Apache Phoenix 下载Phoenix二进制包安装，也可以从Apache或Github下载源码使用Maven编译，命令如下：
　　　　# mvn package -DskipTests -Dhadoop.profile=2

　　将Phoenix服务端二进制拷贝至HBase region server lib 目录下，这里以CDH为例：

　　　　# mv $PHOENIX_HOME/phoenix-core/target/phoenix-core-4.0.0-incubating.jar  /opt/cloudera/parcels/CDH/lib/hbase/lib/
　　　　# mv $PHOENIX_HOME/phoenix-hadoop2-compat/target/phoenix-hadoop2-compat-4.0.0-incubating.jar /opt/cloudera/parcels/CDH/lib/hbase/lib/

　　将Phoenix客户端二进制增加至Client CLASSPATH

　　　　# CLASSPATH=$CLASSPATH:$PHOENIX_HOME/phoenix-assembly/target/phoenix-4.0.0-incubating-client.jar

（三）Quick Start

　　1）Shell 操作

　　启动HBase验证

　　　　# hbase shell
　　　　hbase(main):001:0> list

　　启动Phoenix

　　　　# bin/sqlline.py zk_host

 　　　create table test (mykey integer not null primary key, mycolumn varchar); upsert into test values (1,\'Hello\'); 
　　　 upsert into test values (2,\'World!\'); 
      select * from test;

　　注意：Phoenix使用sqlline作为连接终端，详细命令可参考http://www.hydromatic.net/sqlline/manual.html，如可使用如下命令查看HBase数据库表和表的列属性：

　　　　0: jdbc:phoenix:zk_host> !tables
　　　　0: jdbc:phoenix:zk_host> !columns TEST

　　回到HBase Terminal：

　　　　hbase(main):001:0> list
　　　　hbase(main):002:0> scan \'TEST\'

　　2）Java Client

　　　　新建一个Java Client Source 文件，内容如下：

 　　　import java.sql.Connection; 
　　　 import java.sql.DriverManager; 
　　 　import java.sql.ResultSet; 
　　   import java.sql.SQLException; 
　　   import java.sql.PreparedStatement; 
　　   import java.sql.Statement; 
　　　　public class TestJava { 
　　　　　　public static void main(String[] args) throws SQLException { 
　　　　　　　　Statement stmt = null; 
　　　　　　　　ResultSet rset = null; 
　　　　　　　　Connection con = DriverManager.getConnection("jdbc:phoenix:[zk_host]"); 

　　　　　　　　// 已通过Shell建立 //stmt = con.createStatement(); 
　　　　　　　　//stmt.executeUpdate("create table test (mykey integer not null primary key, mycolumn varchar)"); 
　　　　　　　　//stmt.executeUpdate("upsert into test values (1,\'Hello\')"); 
　　　　　　　　//stmt.executeUpdate("upsert into test values (2,\'World!\')"); 
　　　　　　　　//con.commit(); 

　　　　　　　　PreparedStatement statement = con.prepareStatement("select * from test"); 
　　　　　　　　rset = statement.executeQuery();

　　　　　　　　while (rset.next()) { 
　　　　　　　　　　System.out.println(rset.getString("mycolumn")); 
　　　　　　　　} 

　　　　　　　　statement.close(); 
　　　　　　　　con.close(); 
　　　　　　} 
　　　　}

　　　　编译并执行：

　　　　　　# javac TestJava.java
　　　　　　# java -cp phoenix-assembly/target/phoenix-4.0.0-incubating-client.jar:. TestJava

　　3）数据导入

　　　　3.1 Loading CSV data via PSQL

　　　　　　以Phoenix自带的examples为例，建表、导入数据、查询结果，命令如下：

　　　　　　　　#bin/psql.py zk_host examples/WEB_STAT.sql  examples/WEB_STAT.csv  examples/WEB_STAT_QUERIES.sql

　　　　　　详细用法参考如下：

                 　　    http://phoenix.incubator.apache.org/bulk_dataload.html

　　　　3.2 Loading via MapReduce

　　　　　　3.2.1）建表

　　　　　　　　#bin/psql.py zk_host examples/WEB_STAT.sql

　　　　　　3.2.2）创建待导入HDFS数据文件

　　　　　　　　# sudo -u hdfs hadoop fs -mkdir /user/phoenix/
　　　　　　　　# sudo -u hdfs hadoop fs -put examples/WEB_STAT.csv /user/phoenix/

　　　　　　3.2.3）从HDFS向Phoenix HBase导入数据(以CDH为例)

　　　　　　　　# sudo -u hdfs HADOOP_CLASSPATH=$(hbase classpath) hadoop jar phoenix-assembly/target/phoenix-4.0.0-incubating-client.jar \\　　　　　　　　                                               org.apache.phoenix.mapreduce.CsvBulkLoadTool -zookeeper zk_host --table WEB_STAT --input /user/phoenix/WEB_STAT.csv 

　　　　　　　　或者

　　　　　　　　# sudo -u hdfs HADOOP_CLASSPATH=/opt/cloudera/parcels/CDH-5.0.0-1.cdh5.0.0.p0.47/lib/hbase/hbase-protocol.jar:/etc/hbase/conf hadoop jar \\

　　　　　　　　phoenix-assembly/target/phoenix-4.0.0-incubating-client.jar org.apache.phoenix.mapreduce.CsvBulkLoadTool -zookeeper zk_host --table WEB_STAT \\ 

                  　　   --input /user/phoenix/WEB_STAT.csv

　　4）映射现有HBase表

　　　　在Phoenix中通过CREATE TABLE/CREATE VIEW DDL映射HBase表
　　　　　　phoenix> CREATE TABLE "t1" ( pk VARCHAR PRIMARY KEY, "cf".a VARCHAR, "cf".B VARCHAR, "cf".C VARCHAR ); 
　　　　　　phoenix> select * from "t1";

　　　　在HBase中使用Shell创建一个测试表，并插入若干数据

　　　　　　hbase> put \'t1\', \'row1\', \'cf:a\', \'value1\'
　　　　　　hbase> put \'t1\', \'row1\', \'cf:B\', \'value2\'
　　　　　　hbase> put \'t1\', \'row1\', \'cf:c\', \'value3\'
　　　　　　hbase> scan \'t1\'

　　注意：

　　　　1)在Phoenix中DDL/DML是忽略大小写的，而表名和列名是区分大小写的；

　　　　2)CREATE VIEW风险较小，但是readonly无法新增修改数据；

　　　　3)CREATE TABLE风险较大，一旦表结构与HBase原有表不一致，原表将会被新表覆盖，造成数据丢失；

　　　　4)映射表创建后，遗留数据无法查询，新增修改后的数据才可见

（四）高级进阶

　　1）Secondary Indexing

　　在HBase中每条记录都有一个primary row key，所有的查询过滤器都是基于此项进行。而Phoenix设计了二级索引，使得数据检索不再依赖于primary row key，而更像传统的SQL查询。Phoenix支持Mutable Index和Immutable Index，两种索引适合不同的场景，如Immutable Index非常适合time-series data，数据一旦写入便不再修改，只能追加。
　　　　1.1）Mutable Index

　　　　　　CREATE TABLE my_table (k VARCHAR PRIMARY KEY, v1 VARCHAR, v2 BIGINT);

　　　　　　为列v1查询建立索引

　　　　　　CREATE INDEX my_index ON my_table (v1);

　　　　　　为列v1，v2查询建立索引

　　　　　　CREATE INDEX my_index ON my_table (v1) INCLUDE (v2);

　　　　1.2）Immutable Index

　　　　　　要使用此索引形式，必须在建表时指定IMMUTABLE_ROWS=true，如：

 　　　　　CREATE TABLE my_table (k VARCHAR PRIMARY KEY, v VARCHAR) IMMUTABLE_ROWS=true;

　　　　注意：只有当查询cloumns完全匹配索引时，基于索引的查询才能生效。例如：
　　　　　　create table usertable (id varchar primary key, firstname varchar, lastname varchar); 
　　　　　　create index idx_name on usertable (firstname);

　　　　查询语句：
　　　　　　select id, firstname, lastname from usertable where firstname = \'foo\';
　　　　由于索引只有firstname，而查询cloumns还包括了lastname，因此上述查询索引并未生效，可将索引调整如下：
　　　　　　create index idx_name on usertable (firstname) include (lastname);

　　2）Paged Queries

　　　SELECT * FROM TEST LIMIT 1000;　　

　　   SELECT title, author, isbn, description FROM library WHERE published_date > 2010 AND (title, author, isbn) > (?, ?, ?) ORDER BY title, author, isbn LIMIT 20

3）Sequences

　　创建序列

　　　　CREATE SEQUENCE my_sequence;
　　　　CREATE SEQUENCE my_sequence START WITH -1000;
　　　　CREATE SEQUENCE my_sequence INCREMENT BY 10;
　　　　CREATE SEQUENCE my_schema.my_sequence START 0 CACHE 10;

　　插入数据

　　　　

　　删除序列
　　　　DROP SEQUENCE my_sequence;
　　　　DROP SEQUENCE IF EXISTS my_schema.my_sequence;

　　4）Salted Tables

　　在使用连续的row key时，避免RegionServer出现Hotspotting是HBase使用过程中的通用问题。详细描述及解决方案参考http://blog.sematext.com/2012/04/09/hbasewd-avoid-regionserver-hotspotting-despite-writing-records-with-sequential-keys/。 其问题大致可描述为：HBase按照row key的字典序的形式存储记录，这可以非常快速的通过raw key定位数据，以及数据的startkey和endkey范围。在很多应用场景中，很多时候采用自增或自减的序列作为HBase的row key，比如001，002，003……或者499，498，…… 在这种情况下，下一个序列必须由当前序列和步长计算得出，那么如果Region由很多个RegionServer提供服务，这是没有问题的。但很显然在HBase的架构中，一个Region只有一个RegionServer提供服务。因此，当一个Region达到它预先设置的一个最大范围时，将分裂成为两个较小的Region，并由两个RegionServer接管，无论怎样分裂对Region的写入负载总是落在某一个RegionServer的主机上，若需要全局序列实现自增，显然就失去了集群负载均衡的作用，下图是一个负载情况：

找到了问题原因，那么解决方案就相对容易一下，例如可以salt row keys with a prefix，即为row key做一些添加剂，如增加前缀：

　　new_row_key = (++index % BUCKETS_NUMBER) + original_key

这里简单的根据数据的原始自增或自减的记录或者ID作为index，对需要分片的bucket去余进行划分，得到一个新的row key：

那么HBase集群相对之前的负载将会大有改善，如下图所示：

Phoenix中以Salted Table的方式支持上述接近方案，可以做如下定义，其中SALT_BUCKETS是1~256的自然数：

　　CREATE TABLE table (a_key VARCHAR PRIMARY KEY, a_col VARCHAR) SALT_BUCKETS = 20;

　　5）Configuration and Tuning

　　Apache Phoenix作为HBase的一个连接驱动，在客户端做了很多定制优化。因此，应用程序可以对Client端的hbase-site.xml进行参数调优。详细属性说明参考：http://phoenix.incubator.apache.org/tuning.html
　　这里以bin/psql.py客户端脚本执行调优为例，修改$PHOENIX/phoenix-assembly/target目录下的phoenix-4.0.0-incubating-client.jar的hbase-default.xml文件，增加如下内容以实现将客户端默认缓存从100MB调整为1GB：

　　phoenix.query.maxServerCacheBytes
　　1024000000