HBase新版本Java API
之前没有码全，这次增删改查全乎了，网上有很多例子，自己根据实际在用的收集总结了一下
- 导入的包
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;
- 初始化连接对象
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();
- 创建表、删除表
//创建表名对象
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);
- 向表中单行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);
- 批量插入
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()
- 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);
}
}
- 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);
}
- 删除数据
//删除数据
Delete delete = new Delete(Bytes.toBytes("row02"));
delete.addColumn(Bytes.toBytes("info"), Bytes.toBytes("col2"));
Table table = con.getTable(tn);
table.delete(delete);
- 关闭
//5.关闭
if (admin != null){
admin.close();
}
if(con != null){
con.close();
}
