HBase新版本Java API

HBase新版本Java API

之前没有码全，这次增删改查全乎了，网上有很多例子，自己根据实际在用的收集总结了一下

1. 导入的包

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

2. 初始化连接对象

Admin admin=null;
Connection con=null;
try {
    //1.获得配置文件对象
    Configuration conf=HBaseConfiguration.create();
    //设置配置参数
    conf.set("hbase.zookeeper.quorum", "192.168.52.140");
    //2.建立连接
    con=ConnectionFactory.createConnection(conf);
    //3.获得会话
    admin=con.getAdmin();

3. 创建表、删除表

//创建表名对象
TableName tn=TableName.valueOf("stu");
//a.判断数据库是否存在
if(admin.tableExists(tn)){
    System.out.println("====> 表存在，删除表....");
    //先使表设置为不可编辑，关闭表
    admin.disableTable(tn);
    //删除表
    admin.deleteTable(tn);
    System.out.println("表删除成功.....");
}
//创建表结构对象
HTableDescriptor htd=new HTableDescriptor(tn);
//创建列族结构对象
HColumnDescriptor hcd1=new HColumnDescriptor("fm1");
HColumnDescriptor hcd2=new HColumnDescriptor("fm2");
htd.addFamily(hcd1);
htd.addFamily(hcd2);
//创建表
admin.createTable(htd);

4. 向表中单行put数据

//-------过时代码------------
var hTable = new HTable(hbaseConf, TableName.valueOf(hbaseTable))
val put: Put = new Put(Bytes.toBytes(rowKey))
put.addColumn(Bytes.toBytes("列族"), Bytes.toBytes("列"), Bytes.toBytes("值"))
// 当你进行大量的Put的时候，要确认你的HTable的setAutoFlush是关闭着的。
// 否则的话，每执行一个Put就要想区域服务器发一个请求。
// 否则的话，每执行一个Put就要想区域服务器发一个请求。
// autoFlush = false，要等到写缓冲都填满的时候才会发起请求。
hTable.setAutoFlush(false, false)
hTable.setWriteBufferSize(30 * 1024 * 1024) //5M
//put可以放到List里批量提交
hTable.put(put)
hTable.flushCommits()
//---------------旧api会在将来版本删除------------------

//单个插入
Put put =new Put(Bytes.toBytes("row01"));//参数是行健row01
put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("col1"), Bytes.toBytes("value01"));

//获得表对象
Table table=con.getTable(tn);
table.put(put);

5. 批量插入

Put put01 =new Put(Bytes.toBytes("row02"));//参数是行健
put01.addColumn(Bytes.toBytes("info"), Bytes.toBytes("col2"), Bytes.toBytes("value02"))
.addColumn(Bytes.toBytes("info"), Bytes.toBytes("col3"), Bytes.toBytes("value03"));

Put put02 =new Put(Bytes.toBytes("row03"));//参数是行健
put02.addColumn(Bytes.toBytes("info"), Bytes.toBytes("col4"), Bytes.toBytes("value04"));

List<Put> puts=Arrays.asList(put01,put02);

//获得表对象
Table table=con.getTable(tn);
table.put(puts);

• 还有种批量异步插入方式，效率更高

//第二种插入方式，也可批量插入
//设置缓存1m，当达到1m时数据会自动刷到hbase，替代了hTable.setWriteBufferSize(30 * 1024 * 1024)
val params = new BufferedMutatorParams(TableName.valueOf(hbaseTable)).writeBufferSize(1024 * 1024) //字节
//创建一个批量异步与hbase通信的对象
var mutatorTest: BufferedMutator = connection.getBufferedMutator(params)
val put: Put = new Put(Bytes.toBytes(rowKey))
put.addColumn(Bytes.toBytes("列族"), Bytes.toBytes("列"), Bytes.toBytes("值"))
//向hbase插入数据,达到缓存会自动提交,这里也可以通过传入List<put>的方式批量插入
mutatorTest.mutate(put)
//不用每次put后就调用flush，最后调用就行，这个方法替代了旧api的hTable.setAutoFlush(false, false)
mutator.flush()
mutator.close()
connection.close()

6. scan读取数据

Scan scan=new Scan();
//获得表对象
Table table=con.getTable(tn);
//得到扫描的结果集
ResultScanner rs=table.getScanner(scan);
for(Result result:rs){
    //得到单元格集合
    List<Cell> cs=result.listCells();
    for(Cell cell:cs){
        //取行健
        String rowKey=Bytes.toString(CellUtil.cloneRow(cell));
        //取到时间戳
        long timestamp = cell.getTimestamp();
        //取到族列
        String family = Bytes.toString(CellUtil.cloneFamily(cell));  
        //取到修饰名
        String qualifier  = Bytes.toString(CellUtil.cloneQualifier(cell)); 
        //取到值
        String value = Bytes.toString(CellUtil.cloneValue(cell));  

        System.out.println(" ===> rowKey : " + rowKey + ",  timestamp : " +
        timestamp + ", family : " + family + ", qualifier : " + qualifier + ", value : " + value);
    }
}

7. get读取数据

Get get = new Get(Bytes.toBytes("row01"));
get.addColumn(Bytes.toBytes("info"), Bytes.toBytes("col2"));
Table table = con.getTable(tn);
Result r = table.get(get);
List<Cell> cs = r.listCells();
for (Cell cell : cs) {
    String rowKey = Bytes.toString(CellUtil.cloneRow(cell));  //取行键
    long timestamp = cell.getTimestamp();  //取到时间戳
    String family = Bytes.toString(CellUtil.cloneFamily(cell));  //取到族列
    String qualifier  = Bytes.toString(CellUtil.cloneQualifier(cell));  //取到修饰名
    String value = Bytes.toString(CellUtil.cloneValue(cell));  //取到值

    System.out.println(" ===> rowKey : " + rowKey + ",  timestamp : " + 
    timestamp + ", family : " + family + ", qualifier : " + qualifier + ", value : " + value);
}

8. 删除数据

//删除数据
Delete delete = new Delete(Bytes.toBytes("row02"));
delete.addColumn(Bytes.toBytes("info"), Bytes.toBytes("col2"));
Table table = con.getTable(tn);
table.delete(delete);

9. 关闭

//5.关闭
if (admin != null){
    admin.close();
}
if(con != null){
    con.close();
}