spring boot集成hbase-client
参考上文使用
spring-boot-starter-hbase和RowMapper.
@Autowired
private HbaseTemplate hbaseTemplate;
创建表
/**
* 创建表
* @return
* @throws IOException
*/
public String createTable() throws IOException {
Admin admin = hbaseTemplate.getConnection().getAdmin();
HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(table_name));
hTableDescriptor.addFamily(new HColumnDescriptor(column_family));
if (admin.tableExists(TableName.valueOf(table_name))) {
return "tableExists";
} else {
admin.createTable(hTableDescriptor);
return "ok";
}
}
批量插入数据
/**
* 批量插入数据
* @param i
*/
public void puts(int i) {
List<Mutation> puts = new ArrayList<>();
// 设值
while (i > 0) {
Put put = new Put(Bytes.toBytes(Long.toString(18752038428L - i)));
put.addColumn(Bytes.toBytes(column_family), Bytes.toBytes("name"), Bytes.toBytes("JThink" + i));
put.addColumn(Bytes.toBytes(column_family), Bytes.toBytes("age"), Bytes.toBytes(i));
puts.add(put);
i--;
}
this.hbaseTemplate.saveOrUpdates(table_name, puts);
}
根据rowkey查询数据
/**
* 根据rowkey查询数据
* @param row
* @return
*/
public PeopleDto get(String row) {
PeopleDto dto = this.hbaseTemplate.get(table_name, row, new PeopleRowMapper());
return dto;
}
根据rowkey删除数据
/**
* 根据rowkey删除数据
*/
public void delete(String rk) {
Mutation delete = new Delete(Bytes.toBytes(rk));
this.hbaseTemplate.saveOrUpdate(table_name, delete);
}
批量查询数据
/**
* 区间查找 [startRow, stopRow)
* @param startRow
* @param stopRow
* @return
*/
public List<PeopleDto> query(String startRow, String stopRow) {
Scan scan = new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow));
scan.setCaching(5000);
List<PeopleDto> dtos = this.hbaseTemplate.find(table_name, scan, new PeopleRowMapper());
return dtos;
}
注意查找的结果遵循
左闭右开原则.
过滤
// 要查询的表
HTable table = new HTable(conf, "table1");
// 要查询的字段
Scan scan = new Scan();
scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("a"));
scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("b"));
// where条件
// a = 1
SingleColumnValueFilter a = new SingleColumnValueFilter(Bytes.toBytes("cf"),
Bytes.toBytes("a"), CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(1)));
filterList.addFilter(filter);
// b = 2
SingleColumnValueFilter b = new SingleColumnValueFilter(Bytes.toBytes("cf"),
Bytes.toBytes("b"), CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(2)));
// and
FilterList filterList = new FilterList(Operator.MUST_PASS_ALL, a, b);
scan.setFilter(filterList);