03 通过java代码访问Hbase

前面我们熟悉了Hbase的命令行操作，本节我们将了解如何使用java代码访问Hbase服务。

1 系统、软件以及前提约束

hbase已经安装完毕并且启动
https://www.jianshu.com/p/51fb48f7d1ea

本文中作者服务器的ip是192.168.100.200，请读者根据实际情况设置

下载到相关jar包
上述软件下载链接：https://pan.baidu.com/s/1c_skDYabCRSkS5hRUB6lFQ

2 操作

1 在java访问HDFS的那个项目https://www.jianshu.com/p/386cd966c04f中，加入以下的测试类：

import java.io.IOException; 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; public class HbaseTest2 { public static Configuration configuration; public static Connection connection; public static Admin admin; public static void main(String[] args) throws IOException { createTable("t2", new String[] { "cf1", "cf2" }); // listTables(); // insterRow("t2", "rw1", "cf1", "q1", "val1"); getData("t2", "rw1","cf1", // "q1"); scanData("t2", "rw1", "rw2"); // deleRow("t2","rw1","cf1","q1"); deleteTable("t2"); } // 初始化链接 public static void init() { configuration = HBaseConfiguration.create(); configuration.set("hbase.zookeeper.property.clientPort", "2181"); configuration.set("hbase.zookeeper.quorum", "192.168.100.200"); configuration.set("hbase.master", "192.168.100.200:16000"); System.setProperty("hadoop.home.dir", "C:\\hadoop2.7.2"); try { connection = ConnectionFactory.createConnection(configuration); admin = connection.getAdmin(); } catch (IOException e) { e.printStackTrace(); } } // 关闭连接 public static void close() { try { if (null != admin) admin.close(); if (null != connection) connection.close(); } catch (IOException e) { e.printStackTrace(); } } // 建表 public static void createTable(String tableNmae, String[] cols) throws IOException { init(); TableName tableName = TableName.valueOf(tableNmae); if (admin.tableExists(tableName)) { System.out.println("talbe is exists!"); } else { HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName); for (String col : cols) { HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(col); hTableDescriptor.addFamily(hColumnDescriptor); } admin.createTable(hTableDescriptor); } close(); } // 删表 public static void deleteTable(String tableName) throws IOException { init(); TableName tn = TableName.valueOf(tableName); if (admin.tableExists(tn)) { admin.disableTable(tn); admin.deleteTable(tn); } close(); } // 查看已有表 public static void listTables() throws IOException { init(); HTableDescriptor hTableDescriptors[] = admin.listTables(); for (HTableDescriptor hTableDescriptor : hTableDescriptors) { System.out.println(hTableDescriptor.getNameAsString()); } close(); } // 插入数据 public static void insterRow(String tableName, String rowkey, String colFamily, String col, String val) throws IOException { init(); Table table = connection.getTable(TableName.valueOf(tableName)); Put put = new Put(Bytes.toBytes(rowkey)); put.addColumn(Bytes.toBytes(colFamily), Bytes.toBytes(col), Bytes.toBytes(val)); table.put(put); // 批量插入 /* * List<Put> putList = new ArrayList<Put>(); puts.add(put); * table.put(putList); */ table.close(); close(); } // 删除数据 public static void deleRow(String tableName, String rowkey, String colFamily, String col) throws IOException { init(); Table table = connection.getTable(TableName.valueOf(tableName)); Delete delete = new Delete(Bytes.toBytes(rowkey)); // 删除指定列族 // delete.addFamily(Bytes.toBytes(colFamily)); // 删除指定列 // delete.addColumn(Bytes.toBytes(colFamily),Bytes.toBytes(col)); table.delete(delete); // 批量删除 /* * List<Delete> deleteList = new ArrayList<Delete>(); * deleteList.add(delete); table.delete(deleteList); */ table.close(); close(); } // 根据rowkey查找数据 public static void getData(String tableName, String rowkey, String colFamily, String col) throws IOException { init(); Table table = connection.getTable(TableName.valueOf(tableName)); Get get = new Get(Bytes.toBytes(rowkey)); // 获取指定列族数据 // get.addFamily(Bytes.toBytes(colFamily)); // 获取指定列数据 // get.addColumn(Bytes.toBytes(colFamily),Bytes.toBytes(col)); Result result = table.get(get); showCell(result); table.close(); close(); } // 格式化输出 public static void showCell(Result result) { Cell[] cells = result.rawCells(); for (Cell cell : cells) { System.out.println("RowName:" + new String(CellUtil.cloneRow(cell)) + " "); System.out.println("Timetamp:" + cell.getTimestamp() + " "); System.out.println("column Family:" + new String(CellUtil.cloneFamily(cell)) + " "); System.out.println("row Name:" + new String(CellUtil.cloneQualifier(cell)) + " "); System.out.println("value:" + new String(CellUtil.cloneValue(cell)) + " "); } } // 批量查找数据 public static void scanData(String tableName, String startRow, String stopRow) throws IOException { init(); Table table = connection.getTable(TableName.valueOf(tableName)); Scan scan = new Scan(); // scan.setStartRow(Bytes.toBytes(startRow)); // scan.setStopRow(Bytes.toBytes(stopRow)); ResultScanner resultScanner = table.getScanner(scan); for (Result result : resultScanner) { showCell(result); } table.close(); close(); } }

读者可以根据代码中的注释完成测试。