HBase Java API 基础特性 HBase Client API HBase Java API 示例 引入依赖
1 2 3 4 5 <dependency > <groupId > org.apache.hbase</groupId > <artifactId > hbase-client</artifactId > <version > 2.1.4</version > </dependency >
示例
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 public class HBaseUtils { private static Connection connection; static { Configuration configuration = HBaseConfiguration.create(); configuration.set("hbase.zookeeper.property.clientPort" , "2181" ); configuration.set("hbase.zookeeper.quorum" , "hadoop001" ); try { connection = ConnectionFactory.createConnection(configuration); } catch (IOException e) { e.printStackTrace(); } } public static boolean createTable (String tableName, List<String> columnFamilies) { try { HBaseAdmin admin = (HBaseAdmin) connection.getAdmin(); if (admin.tableExists(TableName.valueOf(tableName))) { return false ; } TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName)); columnFamilies.forEach(columnFamily -> { ColumnFamilyDescriptorBuilder cfDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamily)); cfDescriptorBuilder.setMaxVersions(1 ); ColumnFamilyDescriptor familyDescriptor = cfDescriptorBuilder.build(); tableDescriptor.setColumnFamily(familyDescriptor); }); admin.createTable(tableDescriptor.build()); } catch (IOException e) { e.printStackTrace(); } return true ; } public static boolean deleteTable (String tableName) { try { HBaseAdmin admin = (HBaseAdmin) connection.getAdmin(); admin.disableTable(TableName.valueOf(tableName)); admin.deleteTable(TableName.valueOf(tableName)); } catch (Exception e) { e.printStackTrace(); } return true ; } public static boolean putRow (String tableName, String rowKey, String columnFamilyName, String qualifier, String value) { try { Table table = connection.getTable(TableName.valueOf(tableName)); Put put = new Put (Bytes.toBytes(rowKey)); put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(qualifier), Bytes.toBytes(value)); table.put(put); table.close(); } catch (IOException e) { e.printStackTrace(); } return true ; } public static boolean putRow (String tableName, String rowKey, String columnFamilyName, List<Pair<String, String>> pairList) { try { Table table = connection.getTable(TableName.valueOf(tableName)); Put put = new Put (Bytes.toBytes(rowKey)); pairList.forEach(pair -> put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(pair.getKey()), Bytes.toBytes(pair.getValue()))); table.put(put); table.close(); } catch (IOException e) { e.printStackTrace(); } return true ; } public static Result getRow (String tableName, String rowKey) { try { Table table = connection.getTable(TableName.valueOf(tableName)); Get get = new Get (Bytes.toBytes(rowKey)); return table.get(get); } catch (IOException e) { e.printStackTrace(); } return null ; } public static String getCell (String tableName, String rowKey, String columnFamily, String qualifier) { try { Table table = connection.getTable(TableName.valueOf(tableName)); Get get = new Get (Bytes.toBytes(rowKey)); if (!get.isCheckExistenceOnly()) { get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier)); Result result = table.get(get); byte [] resultValue = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier)); return Bytes.toString(resultValue); } else { return null ; } } catch (IOException e) { e.printStackTrace(); } return null ; } public static ResultScanner getScanner (String tableName) { try { Table table = connection.getTable(TableName.valueOf(tableName)); Scan scan = new Scan (); return table.getScanner(scan); } catch (IOException e) { e.printStackTrace(); } return null ; } public static ResultScanner getScanner (String tableName, FilterList filterList) { try { Table table = connection.getTable(TableName.valueOf(tableName)); Scan scan = new Scan (); scan.setFilter(filterList); return table.getScanner(scan); } catch (IOException e) { e.printStackTrace(); } return null ; } public static ResultScanner getScanner (String tableName, String startRowKey, String endRowKey, FilterList filterList) { try { Table table = connection.getTable(TableName.valueOf(tableName)); Scan scan = new Scan (); scan.withStartRow(Bytes.toBytes(startRowKey)); scan.withStopRow(Bytes.toBytes(endRowKey)); scan.setFilter(filterList); return table.getScanner(scan); } catch (IOException e) { e.printStackTrace(); } return null ; } public static boolean deleteRow (String tableName, String rowKey) { try { Table table = connection.getTable(TableName.valueOf(tableName)); Delete delete = new Delete (Bytes.toBytes(rowKey)); table.delete(delete); } catch (IOException e) { e.printStackTrace(); } return true ; } public static boolean deleteColumn (String tableName, String rowKey, String familyName, String qualifier) { try { Table table = connection.getTable(TableName.valueOf(tableName)); Delete delete = new Delete (Bytes.toBytes(rowKey)); delete.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(qualifier)); table.delete(delete); table.close(); } catch (IOException e) { e.printStackTrace(); } return true ; } }
数据库连接 在上面的代码中,在类加载时就初始化了 Connection 连接,并且之后的方法都是复用这个 Connection,这时我们可能会考虑是否可以使用自定义连接池来获取更好的性能表现?实际上这是没有必要的。
首先官方对于 Connection
的使用说明如下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 Connection Pooling For applications which require high-end multithreaded access (e.g., web-servers or application servers that may serve many application threads in a single JVM), you can pre-create a Connection, as shown in the following example:对于高并发多线程访问的应用程序(例如,在单个 JVM 中存在的为多个线程服务的 Web 服务器或应用程序服务器), 您只需要预先创建一个 Connection。例子如下: // Create a connection to the cluster. Configuration conf = HBaseConfiguration.create (); try (Connection connection = ConnectionFactory.createConnection (conf); Table table = connection.getTable(TableName.valueOf(tablename))) { // use table as needed, the table returned is lightweight }
之所以能这样使用,这是因为 Connection 并不是一个简单的 socket 连接,接口文档 中对 Connection 的表述是:
1 2 3 4 5 6 7 A cluster connection encapsulating lower level individual connections to actual servers and a connection to zookeeper. Connections are instantiated through the ConnectionFactory class .The lifecycle of the connection is managed by the caller, who has to close () the connection to release the resources.Connection 是一个集群连接,封装了与多台服务器(Matser/Region Server )的底层连接以及与 zookeeper 的连接。连接通过 ConnectionFactory 类实例化。连接的生命周期由调用者管理,调用者必须使用 close () 关闭连接以释放资源。
之所以封装这些连接,是因为 HBase 客户端需要连接三个不同的服务角色:
Zookeeper :主要用于获取 meta
表的位置信息,Master 的信息;
HBase Master :主要用于执行 HBaseAdmin 接口的一些操作,例如建表等;
HBase RegionServer :用于读、写数据。
Connection 对象和实际的 Socket 连接之间的对应关系如下图:
在 HBase 客户端代码中,真正对应 Socket 连接的是 RpcConnection
对象。HBase 使用 PoolMap
这种数据结构来存储客户端到 HBase 服务器之间的连接。PoolMap
的内部有一个 ConcurrentHashMap
实例,其 key 是 ConnectionId
(封装了服务器地址和用户 ticket),value 是一个 RpcConnection
对象的资源池。当 HBase 需要连接一个服务器时,首先会根据 ConnectionId
找到对应的连接池,然后从连接池中取出一个连接对象。
1 2 3 4 5 6 7 8 9 10 11 12 @InterfaceAudience.Private public class PoolMap <K, V > implements Map <K, V > { private PoolType poolType; private int poolMaxSize; private Map<K, Pool<V>> pools = new ConcurrentHashMap<>(); public PoolMap(PoolType poolType) { this .poolType = poolType; } .....
HBase 中提供了三种资源池的实现,分别是 Reusable
,RoundRobin
和 ThreadLocal
。具体实现可以通 hbase.client.ipc.pool.type
配置项指定,默认为 Reusable
。连接池的大小也可以通过 hbase.client.ipc.pool.size
配置项指定,默认为 1,即每个 Server 1 个连接。也可以通过修改配置实现:
1 2 3 config .set ("hbase.client.ipc.pool.type" ,...);config .set ("hbase.client.ipc.pool.size" ,...);connection = ConnectionFactory.createConnection(config );
由此可以看出 HBase 中 Connection 类已经实现了对连接的管理功能,所以我们不必在 Connection 上在做额外的管理。
另外,Connection 是线程安全的,但 Table 和 Admin 却不是线程安全的,因此正确的做法是一个进程共用一个 Connection 对象,而在不同的线程中使用单独的 Table 和 Admin 对象。Table 和 Admin 的获取操作 getTable()
和 getAdmin()
都是轻量级,所以不必担心性能的消耗,同时建议在使用完成后显示的调用 close()
方法来关闭它们。
概述 HBase 的主要客户端操作是由 org.apache.hadoop.hbase.client.HTable
提供的。创建 HTable 实例非常耗时,所以,建议每个线程只创建一次 HTable 实例。
HBase 所有修改数据的操作都保证了行级别的原子性。要么读到最新的修改,要么等待系统允许写入改行修改
用户要尽量使用批处理(batch)更新来减少单独操作同一行数据的次数
写操作中设计的列的数目并不会影响该行数据的原子性,行原子性会同时保护到所有列
创建 HTable 实例(指的是在 java 中新建该类),每个实例都要扫描.META. 表,以检查该表是否存在,推荐用户只创建一次 HTable 实例,而且是每个线程创建一个
如果用户需要多个 HTable 实例,建议使用 HTablePool 类(类似连接池)
CRUD 操作 put Table
接口提供了两个 put
方法
1 2 3 4 void put (Put put) throws IOException;void put (List<Put> puts) throws IOException;
Put 类提供了多种构造器方法用来初始化实例。
Put 类还提供了一系列有用的方法:
多个 add
方法:用于添加指定的列数据。
has
方法:用于检查是否存在特定的单元格,而不需要遍历整个集合
getFamilyMap
方法:可以遍历 Put 实例中每一个可用的 KeyValue 实例
getRow 方法:用于获取 rowkey Put.heapSize() 可以计算当前 Put 实例所需的堆大小,既包含其中的数据,也包含内部数据结构所需的空间
KeyValue 类 特定单元格的数据以及坐标,坐标包括行键、列族名、列限定符以及时间戳KeyValue(byte[] row, int roffset, int rlength, byte[] family, int foffoset, int flength, byte[] qualifier, int qoffset, int qlength, long timestamp, Type type, byte[] value, int voffset, int vlength)
每一个字节数组都有一个 offset 参数和一个 length 参数,允许用户提交一个已经存在的字节数组进行字节级别操作。 行目前来说指的是行键,即 Put 构造器里的 row 参数。
客户端的写缓冲区 每一个 put 操作实际上都是一个 RPC 操作,它将客户端数据传送到服务器然后返回。
HBase 的 API 配备了一个客户端的写缓冲区,缓冲区负责收集 put 操作,然后调用 RPC 操作一次性将 put 送往服务器。
1 2 void setAutoFlush (boolean autoFlush) boolean isAutoFlush ()
默认情况下,客户端缓冲区是禁用的。可以通过 table.setAutoFlush(false)
来激活缓冲区。
Put 列表 批量提交 put
列表:
1 void put (List<Put> puts) throws IOException
注意:批量提交可能会有部分修改失败。
原子性操作 compare-and-set checkAndPut
方法提供了 CAS 机制来保证 put 操作的原子性。
get 1 Result get (Get get) throws IOException
1 2 3 4 Get(byte [] row) Get(byte [] row, RowLock rowLock) Get addColumn (byte [] family, byte [] qualifier ) Get addFamily (byte [] family )
Result 类 当用户使用 get()
方法获取数据,HBase 返回的结果包含所有匹配的单元格数据,这些数据被封装在一个 Result
实例中返回给用户。
Result 类提供的方法如下:
1 2 3 4 5 6 7 byte [] getValue(byte [] family, byte [] qualifier)byte [] value()byte [] getRow()int size () boolean isEmpty () KeyValue[] raw() List<KeyValue> list ()
delete 1 void delete (Delete delete ) throws IOException
1 2 Delte(byte [] row) Delete(byte [] row, long timestamp, RowLock rowLock)
1 2 3 4 Delete deleteFamily (byte [] family ) Delete deleteFamily (byte [] family, long timestamp ) Delete deleteColumns (byte [] family, byte [] qualifier ) Delete deleteColumn (byte [] family, byte [] qualifier )
批处理操作 Row 是 Put、Get、Delete 的父类。
1 2 void batch (List<Row> actions, Object[] results) throws IOException, InterruptedExceptionObject batch (List<Row> actions) throws IOException, InterruptedException
行锁 region 服务器提供了行锁特性,这个特性保证了只有一个客户端能获取一行数据相应的锁,同时对该行进行修改。
如果不显示指定锁,服务器会隐式加锁。
扫描 scan,类似数据库系统中的 cursor,利用了 HBase 提供的底层顺序存储的数据结构。
调用 HTable 的 getScanner 就可以返回扫描器
1 2 ResultScanner getScanner (Scan scan) throws IOException ResultScanner getScanner (byte [] family) throws IOException
Scan 类构造器可以有 startRow,区间一般为 [startRow, stopRow)
1 2 Scan(byte [] startRow, Filter filter) Scan(byte [] startRow)
ResultScanner 以行为单位进行返回
1 2 3 Result next () throws IOException Result[] next(int nbRows) throws IOException void close ()
缓存与批量处理 每一个 next()调用都会为每行数据生成一个单独的 RPC 请求
可以设置扫描器缓存
1 2 void setScannerCaching (itn scannerCaching) int getScannerCaching ()
缓存是面向行一级操作,批量是面向列一级操作
1 2 void setBatch (int batch) int getBatch
RPC 请求的次数=(行数*每行列数)/Min(每行的列数,批量大小)/扫描器缓存
各种特性 Bytes
类提供了一系列将原生 Java 类型和字节数组互转的方法。
参考资料