HBase【操作Java api】
2024-09-07 03:46:21
一.导入依赖
创建模块，导入以下依赖，maven默认编译版本是1.5，用1.8编译。
pom.xml
<dependencies>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>2.0.5</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-server</artifactId>
<version>2.0.5</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>2.3.2</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
</plugins>
</build>
二.HBase工具类
package com.bigdata.hbasedemo;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
/**
* @description: TODO HBase操作工具类
* @author: HaoWu
* @create: 2020/7/15 22:51
*/
public class HBaseUtils {
private static Admin admin = null;
private static Configuration configuration = null;
private static Connection connection = null;
/**
* 私有构造方法
*/
private HBaseUtils() {
//创建配置信息并配置
configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.quorum", "hadoop102:2181,hadoop103:2181,hadoop104:2181");
configuration.set("hbase.rootdir", "hdfs:hadoop102:8020/hbase");
try {
//获取HBase连接对象
connection = ConnectionFactory.createConnection(configuration);
//获取Admin对象
admin = connection.getAdmin();
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 私有成员变量，单例模式
*/
private static HBaseUtils instance = null;
public static synchronized HBaseUtils getInstance() {
if (null == instance) {
instance = new HBaseUtils();
}
return instance;
}
/**
* 关闭连接
*/
public static void close() {
try {
if (admin != null) {
admin.close();
}
if (connection != null) {
connection.close();
}
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 根据表名获取Table实例
*
* @param tableName
* @return
*/
public static Table getTable(String nameSpace, String tableName) {
Table table = null;
try {
table = connection.getTable(TableName.valueOf(nameSpace, tableName));
} catch (IOException e) {
e.printStackTrace();
}
return table;
}
/**
* 创建表
*
* @param nameSpace：命名空间
* @param tableName：表名
* @param splitsKey：预分区
* @param columnFamilies：可变列簇，可设置多个
* @return
*/
public static void createTable(String nameSpace, String tableName, byte[][] splitsKey, String... columnFamilies) {
//获取TableName对象
TableName name = TableName.valueOf(nameSpace, tableName);
try {
//判断表是否存在,存在就删除
if (admin.tableExists(name)) {
admin.disableTable(name);
admin.deleteTable(name);
}
//列簇集合描述
ArrayList<ColumnFamilyDescriptor> columnFamilyDescriptors = new ArrayList<>();
for (String columnFamily : columnFamilies) {
ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(columnFamily.getBytes()).build();
columnFamilyDescriptors.add(columnFamilyDescriptor);
}
//设置列簇，获取表描述器
TableDescriptor build = TableDescriptorBuilder.newBuilder(name).setColumnFamilies(columnFamilyDescriptors).build();
admin.createTable(build, splitsKey);
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 修改表结构：删除列簇1，修改列簇2的VERSION为3,新增一个列簇3
*
* @param nameSpace
* @param tableName
* @param columnFamfily1
* @param columnFamfily2
* @param columnFamfily3 TODO 异常处理没做好，图方便，建议自己每个操作单独写方法进行异常处理
*/
public static void alterTable(String nameSpace, String tableName, String columnFamfily1, String columnFamfily2, String columnFamfily3) {
TableName name = TableName.valueOf(nameSpace, tableName);
try {
if (admin.tableExists(name)) {
//1.删除列簇1
admin.deleteColumnFamily(name, Bytes.toBytes(columnFamfily1));
//2.修改列簇2
ColumnFamilyDescriptor build2 = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamfily2)).setMinVersions(4).setMaxVersions(4).build();
admin.modifyColumnFamily(name, build2);
//3.增加列簇3
ColumnFamilyDescriptor build3 = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamfily3)).build();
admin.addColumnFamily(name, build3);
}
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 删除表
*
* @param nameSpace
* @param tableName
*/
public static void deleteTable(String nameSpace, String tableName) {
TableName name = TableName.valueOf(nameSpace, tableName);
try {
//1.禁用表
admin.disableTable(name);
//2.删除表
admin.deleteTable(name);
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 插入一条数据，值为String-单行单列簇单列
*
* @param tableName：表名
* @param timeStamp：时间戳
* @param rowKey：RowKey
* @param columnFamily：列簇名
* @param column：列限定符
* @param value：String类型值
*/
public static void putStringValue(String nameSpace, String tableName, long timeStamp, String rowKey, String columnFamily, String column, String value) {
//1.获取表操作对象
Table table = getTable(nameSpace, tableName);
//2.封装Put对象,包括rowKey，timeStamp，列簇，列，列值
Put put = new Put(Bytes.toBytes(rowKey), timeStamp);
put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
try {
table.put(put);
} catch (IOException e) {
e.printStackTrace();
} finally {
if (table != null) {
try {
table.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
/**
* 插入一条数据，值为int-单行单列簇单列
*
* @param tableName：表名
* @param timeStamp：时间戳
* @param rowKey：RowKey
* @param columnFamily：列簇名
* @param column：列限定符
* @param value：int类型值
*/
public static void putintValue(String nameSpace, String tableName, long timeStamp, String rowKey, String columnFamily, String column, int value) {
//1.获取表操作对象
Table table = getTable(nameSpace, tableName);
//2.封装Put对象,包括rowKey，timeStamp，列簇，列，列值
Put put = new Put(Bytes.toBytes(rowKey), timeStamp);
put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
try {
table.put(put);
} catch (IOException e) {
e.printStackTrace();
} finally {
if (table != null) {
try {
table.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
/**
* 插入多行，多列，单列簇
*
* @param nameSpace
* @param tableName
* @param columnFamily
* @param mapList:将一个Cell数据封装成一个map<String,String>，所有的值类型要求为String。 mapList数据格式：{{ column=name, rowkey=1001, value=lisi},{ column=age, rowkey=1001, value=20}，{ column=address, rowkey=1001, value="BeiJing"}...}
*/
public static void putMultipleRowMultipleColumnOneFamily(String nameSpace, String tableName, String columnFamily, List<Map<String, String>> mapList) {
//1.获取表操作对象
Table table = getTable(nameSpace, tableName);
//2.创建List<Put>对象
List<Put> puts = new ArrayList<>();
for (Map<String, String> map : mapList) {
//3.将每个map数据封装成一个Put对象
Put put = new Put(Bytes.toBytes(map.get("rowkey")));
put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(map.get("column")), Bytes.toBytes(map.get("value")));
puts.add(put);
}
try {
//4.执行put操作
table.put(puts);
} catch (IOException e) {
e.printStackTrace();
} finally {
if (table != null) {
try {
table.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
/**
* 插入单行，多列，单列簇
*
* @param nameSpace
* @param tableName
* @param columnFamily
* @param mapList:将一个Cell数据封装成一个map<String,String>，所有的值类型要求为String。 mapList数据格式：{{ column=name, rowkey=1001, value=lisi},{ column=age, rowkey=1001, value=20}，{ column=address, rowkey=1001, value="BeiJing"}...}
*/
public static void putOneRowMultipleColumnOneFamily(String nameSpace, String tableName, String rowKey, String columnFamily, List<Map<String, String>> mapList) {
Table table = getTable(nameSpace, tableName);
List<Put> puts = new ArrayList<>();
for (Map<String, String> map : mapList) {
Put put = new Put(Bytes.toBytes(rowKey));
put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(map.get("column")), Bytes.toBytes(map.get("value")));
puts.add(put);
}
try {
table.put(puts);
} catch (IOException e) {
e.printStackTrace();
} finally {
if (table != null) {
try {
table.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
/**
*修改数据和新增数据一致
*/
/**
* 删除单行单列簇单列数据
*
* @param nameSpace
* @param tableName
* @param family
* @param column
* @param rowkey
*/
public static void deleteOneRowOneCloumn(String nameSpace, String tableName, String family, String column, String rowkey) {
Table table = getTable(nameSpace, tableName);
Delete delete = new Delete(Bytes.toBytes(rowkey));
delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
try {
table.delete(delete);
} catch (IOException e) {
e.printStackTrace();
} finally {
if (table != null)
try {
table.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 查询单行单列簇-可变列
* @param nameSpace
* @param tableName
* @param family
* @param columns
* @param rowkey
*/
public static void get(String nameSpace, String tableName, String family, String rowkey,String... columns) {
//1.创建table对象
Table table = getTable(nameSpace, tableName);
//2.创建Get对象
Get get = new Get(Bytes.toBytes(rowkey));
//3.添加列描述
for (String column : columns) {
get.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
}
try {
//4.执行查询获取结果
Result result = table.get(get);
//5.获取Cell集合，然后遍历打印
List<Cell> cells = result.listCells();
for (Cell cell : cells) {
//列簇
System.out.println(Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
//列
System.out.println(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
//值
if ("age".equals(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()))) {
//值为int类型
System.out.println(Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
}else {
//值为String类型
System.out.println(Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));}
System.out.println("***************************************************************************");
}
} catch (IOException e) {
e.printStackTrace();
} finally {
if (table != null)
try {
table.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* Scan 扫描表-单列簇
* @param nameSpace
* @param tableName
* @param family
*/
public static void scanTable(String nameSpace, String tableName,String family){
//1.创建table对象
Table table = getTable(nameSpace, tableName);
//2.创建Scan对象
Scan scan = new Scan();
//3.指定列簇
scan.addFamily(Bytes.toBytes(family));
try {
//是否结果缓存
scan.setCacheBlocks(false);
//4.执行扫描，遍历结果
ResultScanner resultScanners = table.getScanner(scan);
Iterator<Result> iterator = resultScanners.iterator();
while (iterator.hasNext()){
Result result = iterator.next();
List<Cell> cells = result.listCells();
for (Cell cell : cells) {
//RowKey
System.out.print(Bytes.toString(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength())+":");
//列
System.out.print(Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength())+":");
//值
if("age".equals(Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()))){
//值为int类型
System.out.print(Bytes.toInt(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength())+":");
}else {
//值为String类型
System.out.print(Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength())+":");
}
System.out.println("***************************************************************************");
}
}
} catch (IOException e) {
e.printStackTrace();
} finally {
if (table != null)
try {
table.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 获取单值比较器:针对单列族单列进行String值过滤{小于|小于等于|等于|不等于|大于等于|大于}
* @param family 列簇
* @param column 列名
* @param op 枚举类{LESS|LESS_OR_EQUAL|EQUAL|NOT_EQUAL|GREATER_OR_EQUAL|GREATER}
* {小于|小于等于|等于|不等于|大于等于|大于}
* @param value 值
* @return
*/
public static SingleColumnValueFilter getStringSingleColumnValueFilter(String family,String column,CompareOperator op,String value){
return new SingleColumnValueFilter(Bytes.toBytes(family), Bytes.toBytes(column), op, Bytes.toBytes(value));
}
/**
* 获取单值比较器:针对单列族单列进行int值过滤{小于|小于等于|等于|不等于|大于等于|大于}
* @param family 列簇
* @param column 列名
* @param op 枚举类{LESS|LESS_OR_EQUAL|EQUAL|NOT_EQUAL|GREATER_OR_EQUAL|GREATER}
* {小于|小于等于|等于|不等于|大于等于|大于}
* @param value 值
* @return
*/
public static SingleColumnValueFilter getIntSingleColumnValueFilter(String family,String column,CompareOperator op,int value){
return new SingleColumnValueFilter(Bytes.toBytes(family), Bytes.toBytes(column), op, Bytes.toBytes(value));
}
/**
* 列簇过滤器
* @param family
*/
public static FamilyFilter getFamilyFilter(String family){
return new FamilyFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes(family)));
}
/**
*列过滤器
* @param column
* @return
*/
public static QualifierFilter getQualifierFilter(String column){
return new QualifierFilter(CompareOperator.EQUAL,new BinaryComparator(Bytes.toBytes(column)));
}
/**
* 针对全表的值：过滤出包含值包含"value"的行
* 在 where name like %value% 结合列簇->列->值过滤达到效果
* @param value
* @return
*/
public static ValueFilter getValueFilter(String value){
return new ValueFilter(CompareOperator.EQUAL, new SubstringComparator(value));
}
/**
* 获取过滤器链：必须全满足 filter1 and filter2 and filter3
* @param filters
* @return
*/
public static FilterList getFilterListALL(Filter... filters){
FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
for (Filter filter : filters) {
filterList.addFilter(filter);
}
return filterList;
}
/**
* 获取过滤器链：最少有一个满足 filter1 or filter2 or filter3
* @param filters
* @return
*/
public static FilterList getFilterListOne(Filter... filters){
FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
for (Filter filter : filters) {
filterList.addFilter(filter);
}
return filterList;
}
}
三.测试类
package com.bigdata.hbasedemo;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
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.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Test;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* @description: HBaseUtils工具测试类
* @author: HaoWu
* @create: 2020/7/16 8:16
*/
public class HBaseUtilsTest {
//获取工具类对象
HBaseUtils hBaseUtils = HBaseUtils.getInstance();
/**
* 测试创建表
*/
@Test
public void createTable() {
String nameSpace = "bigdata";
String tableName = "user6";
byte[][] splitsKey = new byte[][]{
Bytes.toBytes("10"),
Bytes.toBytes("20"),
Bytes.toBytes("30"),
Bytes.toBytes("40")};
String columnFamily1 = "info1";
String columnFamily2 = "info2";
hBaseUtils.createTable(nameSpace, tableName, splitsKey, columnFamily1, columnFamily2);
}
/**
* 测试修改表
*/
@Test
public void alterTable() {
hBaseUtils.alterTable("bigdata", "user", "info1", "info2", "info3");
}
/**
* 删除表
*/
@Test
public void deleteTable(){
hBaseUtils.deleteTable("bigdata","user5");
}
/**
* 插入一条数据,值为String类型
*/
@Test
public void putStringValue() {
hBaseUtils.putStringValue("bigdata", "user", 1221312312352L, "1004", "info2", "name", "lisi");
}
/**
* 插入一条数据,值为int类型
*/
@Test
public void putintValue() {
hBaseUtils.putintValue("bigdata", "user", 1221312312321L, "1004", "info2", "age", 25);
}
/**
* 插入多行，多列，多值，单列簇
*/
@Test
public void putMultipleRowMultipleColumnOneFamily() {
//数据格式：{{ "column"="name", "rowkey"="1001", "value"="lisi"},{ "column"="address", "rowkey"="1001", "value"="BeiJing"}}
List<Map<String, String>> mapList = new ArrayList<>();
ConcurrentHashMap<String, String> map1 = new ConcurrentHashMap<>();
map1.put("column","name");
map1.put("rowkey","1003");
map1.put("value","lisi1");
System.out.println(map1);
ConcurrentHashMap<String, String> map2 = new ConcurrentHashMap<>();
map2.put("column","address");
map2.put("rowkey","1003");
map2.put("value","BeiJing");
ConcurrentHashMap<String, String> map3 = new ConcurrentHashMap<>();
map3.put("column","name");
map3.put("rowkey","1004");
map3.put("value","wangwu1");
mapList.add(map1);
mapList.add(map2);
mapList.add(map3);
System.out.println(mapList.toString());
hBaseUtils.putMultipleRowMultipleColumnOneFamily("bigdata", "user", "info2", mapList);
}
/**
* 删除单行单列簇单列
*/
@Test
public void deleteOnerowOneColumn(){
hBaseUtils.deleteOneRowOneCloumn("bigdata","user","info2","address","1001");
}
/**
* 查询单行单列簇-可变列
*/
@Test
public void get(){
hBaseUtils.get("bigdata","user","info2","1001","age","name");
}
/**
* scan 扫描表-单列簇
*/
@Test
public void scanTable(){
hBaseUtils.scanTable("bigdata","user","info2");
}
/**
* 根据条件过滤:where name="wangwu" or（name="lisi" and age>30)
*
*/
@Test
public void testFilter1() throws IOException {
Table table = hBaseUtils.getTable("bigdata", "user");
Scan scan = new Scan();
//name="lisi"
SingleColumnValueFilter filter1 = hBaseUtils.getStringSingleColumnValueFilter("info2", "name", CompareOperator.EQUAL, "lisi");
//age>30
SingleColumnValueFilter filter2 = hBaseUtils.getIntSingleColumnValueFilter("info2", "age", CompareOperator.GREATER,30);
//name="lisi" and age>30
FilterList filterList = hBaseUtils.getFilterListALL(filter1, filter2);
//name="wangwu"
SingleColumnValueFilter filter3 = hBaseUtils.getStringSingleColumnValueFilter("info2", "name", CompareOperator.EQUAL, "wangwu");
//name="wangwu" or（name="lisi" and age>30)
FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ONE);
filters.addFilter(filter3);
filters.addFilter(filterList);
//设置过滤器
scan.setFilter(filters);
ResultScanner scanner = table.getScanner(scan);
Iterator<Result> iterator = scanner.iterator();
while (iterator.hasNext()){
Result result = iterator.next();
List<Cell> cells = result.listCells();
for (Cell cell : cells) {
//RowKey
System.out.print(Bytes.toString(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength())+":");
//列
System.out.print(Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength())+":");
//值
if("age".equals(Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()))){
//值为int类型
System.out.print(Bytes.toInt(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength())+":");
}else {
//值为String类型
System.out.print(Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength())+":");
}
System.out.println("***************************************************************************");
}
}
}
/**
* 关闭admin， connection
*/
@Test
public void close(){
hBaseUtils.close();
}
}
最新文章
- ASP.NET Core 1.1 简介
- jQuery知识点一 each()和toggleClass()
- 资源list：Github上关于大数据的开源项目、论文等合集
- c# 垮线程调用控件
- javascript之工厂方式定义对象
- 关于delphi PAServer 远程调试DLL文件
- c# 反射列子
- 【计算机视觉领域】常用的 feature 提取方法，feature 提取工具包
- NSArray block用法
- LESS语法备忘
- 在VirtualBox中安装了Ubuntu后，Ubuntu的屏幕分辨率非常小，操作非常不便。通过安装VirtualBox提供的“增强功能组件”，-摘自网络
- C++中实现 time_t, tm 相互转换
- Javascript 查找元素
- Hibernate Error: a different object with the same identifier value was already associated with the session
- git 更新分支的信息
- 被低估的.net(上) - 微软MonkeyFest 2018广州分享会活动回顾
- Python：爬虫之利用Python获取指定网址上的所有图片—Jaosn niu
- Golang包管理工具之govendor的使用
- [JSOI2008]球形空间产生器 （高斯消元）
- 通用的Sql存储过程