29-hadoop-使用phtonenix工具&分页&数据导入

因为rowkey一般有业务逻辑, 所以不可以直接使用rowkey进行分页, startkey, endkey

想要使用SQL语句对Hbase进行查询，需要使用Apache的开源框架Phoenix。

安装

1, 下载phonenix

http://mirrors.cnnic.cn/apache/phoenix/

注意下载版本对应的phonenix

2, 解压

解压后, 将 phoenix-core-4.5.2-HBase-1.1.jar 拷贝至hbae各个节点的lib目录下

3, 重新启动hbae集群, 加载jar包

4, 启动, 

到加压目录下执行命令

bin/sqlline.py 192.168.208.106:2181

操作

1, 查看所有表信息

!tables

2, 创建表

create table person (
id varchar primary key, 
name varchar
);

3, 插入数据

upsert into person values
('111', 'vini');

这儿的upsert, 相当于sql中的额 insert  和update 功能

4, 删除数据

delete from person
where 
id = '222';

5, 查询所有

select * 
from person;

使用phonenix分页 

hbase中, rowkey并非规则排布的, 并且可能包含业务逻辑, 所以分页不能像关系型数据库一样进行, 但使用phonenix可以新增列然后进行分页

1, 准备数据;

CREATE TABLE  TMP_TRAVEL (ROWKEY VARCHAR PRIMARY KEY,INFO.SP VARCHAR,INFO.EP VARCHAR,INFO.ST VARCHAR,INFO.ET VARCHAR);

数据

UPSERT INTO TMP_TRAVEL VALUES('16357298756_20160201112343','北京','西三旗','20160203');
UPSERT INTO TMP_TRAVEL VALUES('16357298756_20160202112343','北京','西三旗','20160203');
UPSERT INTO TMP_TRAVEL VALUES('16357298756_20160203112343','北京','西三旗','20160203');
UPSERT INTO TMP_TRAVEL VALUES('16357298756_20160204112343','北京','西三旗','20160203');
UPSERT INTO TMP_TRAVEL VALUES('16357298756_20160205112343','北京','西三旗','20160203');
UPSERT INTO TMP_TRAVEL VALUES('16357298756_20160206112343','北京','西三旗','20160203');
UPSERT INTO TMP_TRAVEL VALUES('16357298756_20160207112343','北京','西三旗','20160203');
UPSERT INTO TMP_TRAVEL VALUES('16357298756_20160208112343','北京','西三旗','20160203');
UPSERT INTO TMP_TRAVEL VALUES('16357298756_20160209112343','北京','西三旗','20160203');
UPSERT INTO TMP_TRAVEL VALUES('16357298756_20160210112343','北京','西三旗','20160203');

2, 创建新表, 并增加 pageId

CREATE TABLE TRAVEL (ROWKEY VARCHAR PRIMARY KEY,INFO.SP VARCHAR,INFO.EP VARCHAR,INFO.ST VARCHAR,INFO.ET VARCHAR,PAGEID BIGINT);

3, 创建sequence作为pageId的值

CREATE SEQUENCE SEQ;

4, 将原数据拷贝到新表

UPSERT INTO TRAVEL SELECT ROWKEY,SP,EP,ST,ET,NEXT VALUE FOR SEQ FROM TMP_TRAVEL;

5, 使用 where pageId> ? and limit ? 进行分页

6, java代码

import org.junit.Before;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;

/**
 * Created by Administrator on 2016/3/3 0003.
 */
public class HbaseJdbc {

    private JdbcTemplate jdbcTemplate;

    public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
        this.jdbcTemplate = jdbcTemplate;
    }
    @Before
    public void before(){
        ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
        jdbcTemplate = (JdbcTemplate) context.getBean("phoenixJdbcTemplate");
    }
    @Test
    public void test(){
        List<com.sxt.hbase.Travel> pageRecords = findPageRecords(3, 3, "20160202", "163572", "98756");
        for (Travel tt:pageRecords){
            System.out.println(tt.getROWKEY());
        }
    }

    public List<Travel> findPageRecords(int currentPageNum, int pageSize, String ST, String SP, String EP) {


        //第一个参数为SQL语句，第二参数的RowMapper将每一行结果映射成一个Java对象，方便将其他封装到JavaBean中，第三个参数为占位符值（为可变参数）
        List<Travel> travels =  jdbcTemplate.query("SELECT * FROM TRAVEL where PAGEID > ? AND ST >= ? AND ROWKEY LIKE ? limit ?",
                new RowMapper<Travel>() {
                    public Travel mapRow(ResultSet rs, int rowNum)
                            throws SQLException {
                        Travel travel = new Travel();
                        travel.setROWKEY(rs.getString("ROWKEY"));
                        travel.setSP(rs.getString("SP"));
                        travel.setEP(rs.getString("EP"));
                        travel.setST(rs.getString("ST"));
                        travel.setET(rs.getString("ET"));
                        return travel;
                    }
                },(currentPageNum - 1)*pageSize,ST,SP+EP+"%",pageSize);
        return  travels;
    }
}

分页工具

public class Travel {

    private String ROWKEY;
    private String SP;
    private String EP;
    private String ST;
    private String ET;

    public String getROWKEY() {
        return ROWKEY;
    }

    public void setROWKEY(String ROWKEY) {
        this.ROWKEY = ROWKEY;
    }

    public String getSP() {
        return SP;
    }

    public void setSP(String SP) {
        this.SP = SP;
    }

    public String getEP() {
        return EP;
    }

    public void setEP(String EP) {
        this.EP = EP;
    }

    public String getST() {
        return ST;
    }

    public void setST(String ST) {
        this.ST = ST;
    }

    public String getET() {
        return ET;
    }

    public void setET(String ET) {
        this.ET = ET;
    }
}

数据导入

在上面的实例中, 进行数据导入使用的upsert语句, 非常繁琐, phonenix支持文件导入的方式

psql.py 192.168.208.106:2181 ./WEB-STAT.sql ./WEB_STAT.csv

其中, .sql是建表语句, .csv是需要导入的数据

系列来自尚学堂视频