HBase学习-基本Java API

创建Connection对象

//Connection对象相当于与HBase的一个连接
public static Connection getConnection() throws IOException {
Configuration config = HBaseConfiguration.create();
config.addResource(new Path(System.getenv("HBASE_HOME"),"conf/hbase-site.xml"));
return ConnectionFactory.createConnection(config);
}

创建表

public static void createTable(Connection connection, String tableName,
String[] columnNames) throws IOException {
try (Admin admin = connection.getAdmin()) {
HTableDescriptor table = new HTableDescriptor(TableName.valueOf(tableName));
for (String column : columnNames)
table.addFamily(new HColumnDescriptor(column));
admin.createTable(table);
System.out.println("created table:"+tableName+",with columns:"+Arrays.asList(columnNames));
}
}

删除表

public static void deleteTable(Connection connection, String tableName) throws IOException {
try (Admin admin = connection.getAdmin()) {
admin.disableTable(TableName.valueOf(tableName));
admin.deleteTable(TableName.valueOf(tableName));
System.out.println("deleted table:"+tableName);
}
}

插入put

public static void put(Connection connection, String tableName,String rowKey,
String columnFamily,String column,String value) throws IOException {
try (Table table = connection.getTable(TableName.valueOf(tableName))) {
Put row=new Put(rowKey.getBytes());
row.addColumn(columnFamily.getBytes(), column.getBytes(), value.getBytes());
table.put(row);
System.out.println("put ["+rowKey+","+columnFamily+":"+column+"]="+value+" into table"+tableName);
}
}

查询get

public static Result get(Connection connection, String tableName,String rowKey,
String columnFamily) throws IOException {
return get(connection, tableName, rowKey,columnFamily, null);
}
public static Result get(Connection connection, String tableName,String rowKey,
String columnFamily,String column) throws IOException {
try (Table table = connection.getTable(TableName.valueOf(tableName))) {
Get row=new Get(rowKey.getBytes());
row.addFamily(columnFamily.getBytes());
if(column!=null)
row.addColumn(columnFamily.getBytes(), column.getBytes());
Result result=table.get(row);
for(Cell cell:result.listCells()){
String col=new String(cell.getQualifier());
String value=new String(cell.getValue());
System.out.println("["+rowKey+","+columnFamily+":"+col+"]="+value);
}
return result;
}
}

扫描scan

public static void scan(Connection connection, String tableName,String startRow,String endRow,Filter filter) throws IOException {
try (Table table = connection.getTable(TableName.valueOf(tableName))) {
Scan scan=new Scan();
if(filter!=null)
scan.setFilter(filter);
if(startRow!=null)
scan.setStartRow(startRow.getBytes());
if(endRow!=null)
scan.setStopRow(endRow.getBytes());
ResultScanner scanner=table.getScanner(scan);
for(Result result:scanner){
String rowKey=new String(result.getRow());
for(Cell cell:result.listCells()){
String family=new String(cell.getFamily());
String col=new String(cell.getQualifier());
String value=new String(cell.getValue());
System.out.println("["+rowKey+","+family+":"+col+"]="+value);
}
}
}
}
public static void scan(Connection connection, String tableName) throws IOException {
scan(connection,tableName,null,null,null);
}
public static void scan(Connection connection, String tableName,String startRow,String endRow) throws IOException {
scan(connection,tableName,startRow,endRow,null);
}
public static void scan(Connection connection, String tableName,Filter filter) throws IOException {
scan(connection,tableName,null,null,filter);
}

incr

public static void incr(Connection connection, String tableName,String rowKey,
String columnFamily,String column,long value) throws IOException {
try (Table table = connection.getTable(TableName.valueOf(tableName))) {
Increment incr=new Increment(rowKey.getBytes());
incr.addColumn(columnFamily.getBytes(), column.getBytes(), value);
Result result=table.increment(incr);
for (KeyValue kv : result.raw()) {
System.out.println("KV: " + kv +" Value: " + Bytes.toLong(kv.getValue()));
}
}
}

测试

public static void main(String[] args) throws IOException {
String tableName = "testTable";
String columnFamily_info="info";
String columnFamily_score="score";
String column1_name="name",column1_age="age";
String column2_Math="Math",column2_English="English";
try(Connection conn = getConnection();){
System.out.println("------------------- create table -------------------");
createTable(conn,tableName,new String[]{columnFamily_info,columnFamily_score});

System.out.println("------------------- put -------------------");
put(conn,tableName,"row1",columnFamily_info,column1_name,"sam");
put(conn,tableName,"row1",columnFamily_score,column2_Math,"70");
put(conn,tableName,"row1",columnFamily_score,column2_English,"75");
put(conn,tableName,"row2",columnFamily_score,column2_English,"80");
put(conn,tableName,"row3",columnFamily_info,column1_age,"30");

System.out.println("------------------- get -------------------");
get(conn,tableName,"row1",columnFamily_score,column2_Math);
get(conn,tableName,"row1",columnFamily_score);//get all columns in columnFamily2

System.out.println("------------------- scan -------------------");
scan(conn,tableName,"row1","row3"); //[row1,row3)
System.out.println("------------------- scan all table -------------------");
scan(conn,tableName); //[row1,row3)
System.out.println("------------------- scan with filter -------------------");
Filter filter=new ColumnPrefixFilter("Eng".getBytes());//过滤列族以Eng开头的
scan(conn,tableName,filter); //[row1,row3)

System.out.println("------------------- incr -------------------");
incr(conn,tableName,"row3",columnFamily_score,column2_Math,80);
incr(conn,tableName,"row3",columnFamily_score,column2_Math,-5);

System.out.println("------------------- delete table -------------------");
deleteTable(conn,tableName);
}