Hbase操作工具类

org.apache.hbasehbase-client2.0.5

package javax.utils;import java.io.IOException;
import java.util.ArrayList;
import java.util.List;import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
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;/*** Hbase 操作工具类* * &＃64;author Logan* &＃64;version 1.0.0* &＃64;createDate 2019-05-03**/
public class HbaseUtils {// &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;Common&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;/*** 根据表名获取Table对象* * &＃64;param name 表名&＃xff0c;必要时可指定命名空间&＃xff0c;比如&＃xff1a;“default:user”* &＃64;return Hbase Table 对象* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static Table getTable(String name) throws IOException {TableName tableName &＃61; TableName.valueOf(name);Connection connection &＃61; ConnectionFactory.createConnection();return connection.getTable(tableName);}// &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; Put &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;/*** 根据rowKey生成一个Put对象* * &＃64;param rowKey rowKey* &＃64;return Put对象*/public static Put createPut(String rowKey) {return new Put(Bytes.toBytes(rowKey));}/*** 在Put对象上增加Cell* * &＃64;param put Put对象* &＃64;param cell cell对象* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static void addCellOnPut(Put put, Cell cell) throws IOException {put.add(cell);}/*** 在Put对象上增加值* * &＃64;param put Put对象* &＃64;param family 列簇* &＃64;param qualifier 列* &＃64;param value 字符串类型的值*/public static void addValueOnPut(Put put, String family, String qualifier, String value) {addValueOnPut(put, family, qualifier, Bytes.toBytes(value));}/*** 在Put对象上增加值* * &＃64;param put Put对象* &＃64;param family 列簇* &＃64;param qualifier 列* &＃64;param value 字节数组类型的值&＃xff0c;可以是任意对象序列化而成*/public static void addValueOnPut(Put put, String family, String qualifier, byte[] value) {put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), value);}/*** 在Put对象上增加值* * &＃64;param put Put对象* &＃64;param family 列簇* &＃64;param qualifier 列* &＃64;param ts Timestamp时间戳* &＃64;param value 字符串类型的值*/public static void addValueOnPut(Put put, String family, String qualifier, long ts, String value) {addValueOnPut(put, family, qualifier, ts, Bytes.toBytes(value));}/*** 在Put对象上增加值* * &＃64;param put Put对象* &＃64;param family 列簇* &＃64;param qualifier 列* &＃64;param ts Timestamp时间戳* &＃64;param value 字节数组类型的值&＃xff0c;可以是任意对象序列化而成*/public static void addValueOnPut(Put put, String family, String qualifier, long ts, byte[] value) {put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), ts, value);}/*** 按表名插入一个Put对象包含的数据* * &＃64;param tableName 表名&＃xff0c;必要时可指定命名空间&＃xff0c;比如&＃xff1a;“default:user”* &＃64;param put 要插入的数据对象* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static void put(String tableName, Put put) throws IOException {try (Table table &＃61; getTable(tableName);) {table.put(put);}}/*** 按表名批量插入Put对象包含的数据* * &＃64;param tableName 表名&＃xff0c;必要时可指定命名空间&＃xff0c;比如&＃xff1a;“default:user”* &＃64;param puts 要插入的数据对象集合* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static void put(String tableName, List puts) throws IOException {try (Table table &＃61; getTable(tableName);) {table.put(puts);}}// &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; Get &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;/*** 根据rowKey生成一个查询的Get对象* * &＃64;param rowKey rowKey* &＃64;return Get 对象*/public static Get createGet(String rowKey) {return new Get(Bytes.toBytes(rowKey));}/*** 对查询的Get对象增加指定列簇* * &＃64;param get* &＃64;param family*/public static void addFamilyOnGet(Get get, String family) {get.addFamily(Bytes.toBytes(family));}/*** 对查询的Get对象增加指定列簇和列* * &＃64;param get* &＃64;param family* &＃64;param qualifier*/public static void addColumnOnGet(Get get, String family, String qualifier) {get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));}/*** 根据表名和rowKey查询结果&＃xff08;包含全部列簇和列&＃xff09;* * &＃64;param tableName 表名&＃xff0c;必要时可指定命名空间&＃xff0c;比如&＃xff1a;“default:user”* &＃64;param rowKey 查询rowKey* &＃64;return 查询结果Result* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static Result get(String tableName, String rowKey) throws IOException {Get get &＃61; createGet(rowKey);return get(tableName, get);}/*** 根据表名和rowKey数组批量查询结果&＃xff08;包含全部列簇和列&＃xff09;* * &＃64;param tableName 表名&＃xff0c;必要时可指定命名空间&＃xff0c;比如&＃xff1a;“default:user”* &＃64;param rowKeys 查询rowKey数组* &＃64;return 查询结果Result数组* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static Result[] get(String tableName, String[] rowKeys) throws IOException {List gets &＃61; new ArrayList();for (String rowKey : rowKeys) {gets.add(createGet(rowKey));}return get(tableName, gets);}/*** 根据表名和Get对象查询结果* * &＃64;param tableName 表名&＃xff0c;必要时可指定命名空间&＃xff0c;比如&＃xff1a;“default:user”* &＃64;param get Hbase查询对象* &＃64;return 查询结果Result* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static Result get(String tableName, Get get) throws IOException {try (Table table &＃61; getTable(tableName);) {return table.get(get);}}/*** 根据表名和Get对象数组查询结果* * &＃64;param tableName 表名&＃xff0c;必要时可指定命名空间&＃xff0c;比如&＃xff1a;“default:user”* &＃64;param gets 多个Hbase查询对象组成的数组* &＃64;return 查询结果Result数组* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static Result[] get(String tableName, List gets) throws IOException {try (Table table &＃61; getTable(tableName);) {return table.get(gets);}}// &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; Scan &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;/*** 根据startRow和stopRow创建扫描对象* * &＃64;param startRow 扫描开始行&＃xff0c;结果包含该行* &＃64;param stopRow 扫描结束行&＃xff0c;结果不包含该行* &＃64;return Scan对象*/public static Scan createScan(String startRow, String stopRow) {Scan scan &＃61; new Scan();scan.withStartRow(Bytes.toBytes(startRow));scan.withStopRow(Bytes.toBytes(stopRow));return scan;}/*** 对扫描对象设置列簇* * &＃64;param scan 扫描对象* &＃64;param family 列簇*/public static void addFamilyOnScan(Scan scan, String family) {scan.addFamily(Bytes.toBytes(family));}/*** 对扫描对象设置列* * &＃64;param scan 扫描对象* &＃64;param family 列簇* &＃64;param qualifier 列簇下对应的列*/public static void addColumnOnScan(Scan scan, String family, String qualifier) {scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));}/*** 根据表名和扫描对象扫描数据* * &＃64;param tableName 表名&＃xff0c;必要时可指定命名空间&＃xff0c;比如&＃xff1a;“default:user”* &＃64;param scan 扫描对象* &＃64;return 扫描结果集对象ResultScanner* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static ResultScanner scan(String tableName, Scan scan) throws IOException {try (Table table &＃61; getTable(tableName);) {return table.getScanner(scan);}}/*** 根据表名、开始行和结束行扫描数据&＃xff08;结果包含开始行&＃xff0c;不包含结束行&＃xff0c;半开半闭区间[startRow, stopRow)&＃xff09;* * &＃64;param tableName 表名&＃xff0c;必要时可指定命名空间&＃xff0c;比如&＃xff1a;“default:user”* &＃64;param startRow 扫描开始行* &＃64;param stopRow 扫描结束行* &＃64;return 扫描结果集对象ResultScanner* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static ResultScanner scan(String tableName, String startRow, String stopRow) throws IOException {return scan(tableName, createScan(startRow, stopRow));}// &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61; Delete &＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;&＃61;/*** 根据rowKey生成一个查询的Delete对象* * &＃64;param rowKey rowKey* &＃64;return Delete对象*/public static Delete createDelete(String rowKey) {return new Delete(Bytes.toBytes(rowKey));}/*** 在Delete对象上增加Cell* * &＃64;param delete Delete对象* &＃64;param cell cell对象* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static void addCellOnDelete(Delete delete, Cell cell) throws IOException {delete.add(cell);}/*** 对删除对象增加指定列簇* * &＃64;param delete Delete对象* &＃64;param family 列簇*/public static void addFamilyOnDelete(Delete delete, String family) {delete.addFamily(Bytes.toBytes(family));}/*** 对删除对象增加指定列簇和列* * &＃64;param delete Delete对象* &＃64;param family 列簇* &＃64;param qualifier 列*/public static void addColumnOnDelete(Delete delete, String family, String qualifier) {delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));}/*** 按表名删除一个Delete对象指定的数据* * &＃64;param tableName 表名&＃xff0c;必要时可指定命名空间&＃xff0c;比如&＃xff1a;“default:user”* &＃64;param delete Delete对象* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static void delete(String tableName, Delete delete) throws IOException {try (Table table &＃61; getTable(tableName);) {table.delete(delete);}}/*** 按表名批量删除Delete对象集合包含的指定数据* * &＃64;param tableName 表名&＃xff0c;必要时可指定命名空间&＃xff0c;比如&＃xff1a;“default:user”* &＃64;param deletes Delete对象集合* &＃64;throws IOException 有异常抛出&＃xff0c;由调用者捕获处理*/public static void delete(String tableName, List deletes) throws IOException {try (Table table &＃61; getTable(tableName);) {table.delete(deletes);}}}

package com.java.demo;import java.util.ArrayList;
import java.util.List;import javax.utils.HbaseUtils;import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
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.util.Bytes;
import org.junit.Test;/*** Hbase 客户端测试* * &＃64;author Logan* &＃64;version 1.0.0* &＃64;createDate 2019-05-03**/
public class HbaseClientDemo {/*** 向user表中插入数据*/&＃64;Testpublic void put() {String tableName &＃61; "default:user";try {List puts &＃61; new ArrayList();Put put &＃61; HbaseUtils.createPut("key1005");HbaseUtils.addValueOnPut(put, "info", "name", "孙悟空");HbaseUtils.addValueOnPut(put, "info", "age", "500");HbaseUtils.addValueOnPut(put, "info", "address", "花果山");// HbaseUtils.put(tableName, put);
puts.add(put);put &＃61; HbaseUtils.createPut("key1006");HbaseUtils.addValueOnPut(put, "info", "name", "沙悟净");HbaseUtils.addValueOnPut(put, "info", "age", "1000");HbaseUtils.addValueOnPut(put, "info", "address", "流沙河");puts.add(put);HbaseUtils.put(tableName, puts);} catch (Exception e) {e.printStackTrace();}}/*** 按rowKey批量查询user表中全部列簇全部列的值*/&＃64;Testpublic void getAllFamily() {try {String tableName &＃61; "default:user";String[] rowKeys &＃61; { "key1001", "key1002", "key1003", "key1005", "key1006" };// 按表名和rowKey查询所有列Result[] results &＃61; HbaseUtils.get(tableName, rowKeys);for (Result result : results) {// 打印查询结果
printResult(result);}} catch (Exception e) {e.printStackTrace();}}/*** 按rowKey查询user表中指定列簇指定列的值*/&＃64;Testpublic void get() {try {String tableName &＃61; "default:user";String rowKey &＃61; "key1002";Get get &＃61; HbaseUtils.createGet(rowKey);HbaseUtils.addColumnOnGet(get, "info", "name");HbaseUtils.addColumnOnGet(get, "info", "age");// 不存在的列&＃xff0c;查询结果不显示HbaseUtils.addColumnOnGet(get, "info", "address");// 如果在增加列后增加已有的列簇&＃xff0c;会返回该列簇的全部列数据&＃xff0c;覆盖前边的增加列// HbaseUtils.addFamilyOnGet(get, "info");
Result result &＃61; HbaseUtils.get(tableName, get);printResult(result);} catch (Exception e) {e.printStackTrace();}}&＃64;Testpublic void scan() {try {String tableName &＃61; "default:user";String startRow &＃61; "key1001";String stopRow &＃61; "key1006";ResultScanner resultScanner &＃61; HbaseUtils.scan(tableName, startRow, stopRow);for (Result result : resultScanner) {printResult(result);}} catch (Exception e) {e.printStackTrace();}}/*** 打印查询结果* * &＃64;param result 查询结果对象*/private void printResult(Result result) {Cell[] cells &＃61; result.rawCells();// 从Result中读取 rowKey
System.out.println(Bytes.toString(result.getRow()));String print &＃61; "%s\t %s:%s \t %s";for (Cell cell : cells) {// 从Cell中取rowKeyString row &＃61; Bytes.toString(CellUtil.cloneRow(cell));String family &＃61; Bytes.toString(CellUtil.cloneFamily(cell));String qualifier &＃61; Bytes.toString(CellUtil.cloneQualifier(cell));String value &＃61; Bytes.toString(CellUtil.cloneValue(cell));System.out.println(String.format(print, row, family, qualifier, value));}}/*** 删除指定列*/&＃64;Testpublic void deleteColumn() {try {String tableName &＃61; "default:user";List deletes &＃61; new ArrayList();Delete delete &＃61; HbaseUtils.createDelete("key1005");HbaseUtils.addColumnOnDelete(delete, "info", "age");HbaseUtils.addColumnOnDelete(delete, "info", "address");// HbaseUtils.delete(tableName, delete);
deletes.add(delete);delete &＃61; HbaseUtils.createDelete("key1006");HbaseUtils.addColumnOnDelete(delete, "info", "address");deletes.add(delete);HbaseUtils.delete(tableName, deletes);} catch (Exception e) {e.printStackTrace();}}/*** 删除指定列簇*/&＃64;Testpublic void deleteFamily() {try {String tableName &＃61; "default:user";List deletes &＃61; new ArrayList();Delete delete &＃61; HbaseUtils.createDelete("key1005");HbaseUtils.addFamilyOnDelete(delete, "info");// HbaseUtils.delete(tableName, delete);
deletes.add(delete);delete &＃61; HbaseUtils.createDelete("key1006");HbaseUtils.addFamilyOnDelete(delete, "info");deletes.add(delete);HbaseUtils.delete(tableName, deletes);} catch (Exception e) {e.printStackTrace();}}}