1、前言
大家在开发Flink的时候，很多时候会把某些固定的维度信息存储在Hbase或者Redis等这些第三方库里，已方便来跟流发生关联查询输出。本文将从如何本地搭建一个Hbase环境开始讲起，到如何构建一个Hbase公共调用类，到如何构建一个异步调用Hbase的公共代码，再到最后实际调用代码后的输出。
2、本地利用Docker搭建HBase环境
2.1、下载Hbase镜像
docker pull harisekhon/hbase2.2、启动Hbase镜像：
docker run -d -h myhbase -p 2181:2181 -p 8080:8080 -p 8085:8085 -p 9090:9090 -p 9095:9095 -p 16000:16000 -p 16010:16010 -p 16020:16020 -p 16201:16201 -p 16301:16301 --name hbase harisekhon/hbaseHbase 端口映射图：
2.3、本地添加host
打开/etc/hosts ，如下图所示，添加一行
127.0.0.1 myhbase
此时在浏览器中输入hbase访问的地址，可以看到hbase的页面
http://localhost:16010/master-status
?3、访问Hbase
3.1、图形化工具访问Hbase
如下的图形化工具，Mac和Windows都可以访问。
?如果需要图形化工具的，评论区留下邮箱，本人看到了会第一时间发送
3.2、Java工具类访问
直接上代码，可以直接使用
package com.horse.utils.hbase;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.CaseFormat;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.util.StringUtils;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
import java.util.List;
/**
* @author ：aaronChen
* @date ：Created in 2022-02-22 16:51
* @description ：
* @modifiedBy ：
*/
@Slf4j
public class HBaseUtils {
public static Connection getHBaseConnection() {
try {
// 1.获取配置文件信息
Configuration configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.quorum", "localhost");
// configuration.set("hbase.zookeeper.property.clientPort", "2181");//可写可不写,默认为2181
// 2.创建连接对象
return ConnectionFactory.createConnection(configuration);
} catch (IOException e) {
log.error("获取Hbase相关基础信息失");
e.printStackTrace();
}
return null;
}
// static {
// try {
// // 1.获取配置文件信息
// Configuration configuration = HBaseConfiguration.create();
// configuration.set("hbase.zookeeper.quorum", "localhost");
// configuration.set("hbase.zookeeper.property.clientPort", "2181");//可写可不写,默认为2181
//
// // 2.创建连接对象
// connection = ConnectionFactory.createConnection(configuration);
//
// // 3.创建Admin对象
// admin = connection.getAdmin();
//
// } catch (IOException e) {
// log.error("获取Hbase相关基础信息失");
// e.printStackTrace();
// }
// }
/**
* @param tableName
* @return true or false
* @throws IOException
* @author aaronChen
* @description 判断表是否存在
*/
public static boolean isTableExist(String tableName, Admin admin) throws IOException {
// 1.判断表是否存在
boolean exists = admin.tableExists(TableName.valueOf(tableName));
// 2.返回结果
return exists;
}
/**
* @param tableName 需要创建的hbase表名,格式必须要是 命名空间:tablename
* @param cfs 可以同时传入多个列簇
* @param createNameSpaceIfExists 如果命名空间不存在，是否创建
* @throws IOException
* @author aaronChen
* @description 创建表
*/
public static boolean createTable(Admin admin, String tableName, boolean createNameSpaceIfExists, String... cfs) throws IOException {
//1、判断传入的表是否正确
if (tableName.split(":").length != 2) {
log.error("{},传入的表名有问题，需要传入 nameSpace:tableName 这种格式", tableName);
return false;
}
String nameSpace = tableName.split(":")[0];
//2、判断命名空间相关信息
if (!isNameSpaceExist(admin, nameSpace)) {
if (createNameSpaceIfExists) {
log.info("{},该命名空间不存在，开始创建该命名空间", nameSpace);
createNameSpace(admin, nameSpace);
} else {
log.error("{},该命名空间不存在", nameSpace);
return false;
}
}
// 3、判断表是否存在
if (isTableExist(tableName, admin)) {
log.error("{},表已存在", tableName);
return false;
}
// 4、判断是否存在列族信息
if (cfs.length <= 0) {
log.info("请设置列族信息!");
return false;
}
// 5、创建表描述器
try {
if (admin.tableExists(TableName.valueOf(tableName))) {
return false;
}
//定义表描述对象
TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
//遍历列族数组
for (String cf : cfs) {
//定义列族描述对象
ColumnFamilyDescriptor columnFamily = ColumnFamilyDescriptorBuilder.of(cf);
//给表添加列族信息
tableDescriptorBuilder.setColumnFamily(columnFamily);
}
//创建表
admin.createTable(tableDescriptorBuilder.build());
} catch (Exception e) {
e.printStackTrace();
return false;
}
log.info("{},表新建成功", tableName);
return true;
}
/**
* @param tableName
* @throws IOException
* @author aaronChen
* @description 删除表
*/
public static void dropTable(Admin admin, String tableName) throws IOException {
// 1.判断表是否存在
if (!isTableExist(tableName, admin)) {
log.error("{},表不存在", tableName);
return;
}
// 2.使表下线
admin.disableTable(TableName.valueOf(tableName));
// 3.删除表
admin.deleteTable(TableName.valueOf(tableName));
}
/**
* @param nameSpace
* @return true or false
* @throws IOException
* @author aaronChen
* @description 判断一个命名空间是否存在
*/
public static boolean isNameSpaceExist(Admin admin, String nameSpace) throws IOException {
NamespaceDescriptor[] namespaceDescriptors = admin.listNamespaceDescriptors();
for (NamespaceDescriptor namespaceDescriptor : namespaceDescriptors) {
if (namespaceDescriptor.getName().equals(nameSpace)) {
return true;
}
}
return false;
}
/**
* @param namespace
* @author aaronChen
* @description 创建命名空间
*/
public static void createNameSpace(Admin admin, String namespace) {
// 1.创建命名空间描述器
NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(namespace).build();
// 2.创建命名空间
try {
admin.createNamespace(namespaceDescriptor);
} catch (NamespaceExistException e) {
log.error("{},命名空间已存在", namespace);
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* @param tableName
* @param rowKey
* @param columnFamily
* @param columnName
* @param value
* @throws IOException
* @author aaronChen
* @description 向表中插入数据
*/
public static void putData(Connection connection, String tableName, String rowKey, String columnFamily, String columnName, String value) throws IOException {
// 1.获取表对象
Table table = connection.getTable(TableName.valueOf(tableName));
// 2.创建put对象
Put put = new Put(Bytes.toBytes(rowKey));
// 3.给Put对象赋值
put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName), Bytes.toBytes(value));
// 4.插入数据
table.put(put);
// 5.关闭表连接
table.close();
}
/**
* @param tableName
* @param clz
* @param <T>
* @return
* @throws Exception
*/
public static <T> T getData(Connection connection, String tableName, String rowKey, String columnFamily, String columnName, boolean underScoreToCamel, Class<T> clz) throws Exception {
//1、判断传入的参数是否为空或空字符串
if (StringUtils.isNullOrWhitespaceOnly(tableName)) {
log.error("传入的tableName为空:{}", tableName);
return null;
}
if (StringUtils.isNullOrWhitespaceOnly(rowKey)) {
log.error("传入的rowKey为空:{}", rowKey);
return null;
}
if (StringUtils.isNullOrWhitespaceOnly(columnFamily)) {
log.error("传入的columnFamily为空:{}", columnFamily);
return null;
}
Get get = new Get(Bytes.toBytes(rowKey));
if (StringUtils.isNullOrWhitespaceOnly(columnName)) {
get.addFamily(Bytes.toBytes(columnFamily));
// log.info("传入的columnName为空:{},将根据传入的其他信息查询全量的列信息", columnName);
} else {
//传入的不为空的话，就查询传入的值
get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(columnName));
// log.info("传入的columnName为:{},将根据传入的值查询信息", columnName);
}
//4、获取表对象
Table table = connection.getTable(TableName.valueOf(tableName));
//5、设置获取数据的版本数
//get.setMaxVersions(5);
//5、获取数据
Result resultData = table.get(get);
//6、关闭表连接
table.close();
//创建 泛型对象
T t = clz.newInstance();
//给泛型对象赋值
List<Cell> cells = resultData.listCells();
for (Cell cell : cells) {
String resultColumnName = Bytes.toString(CellUtil.cloneQualifier(cell));
String resultValue = Bytes.toString(CellUtil.cloneValue(cell));
//判断是否需要转换为驼峰命名
if (underScoreToCamel) {
if (resultColumnName.contains("_")) {
resultColumnName = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, resultColumnName.toLowerCase());
} else if (resultColumnName.contains("-")) {
resultColumnName = CaseFormat.LOWER_HYPHEN.to(CaseFormat.LOWER_CAMEL, resultColumnName.toLowerCase());
} else {
resultColumnName = CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_CAMEL, resultColumnName);
}
}
BeanUtils.setProperty(t, resultColumnName, resultValue);
}
return t;
}
/**
* @param tableName
* @return 返回一个 JSONArray
* <p>
* [
* {
* "rowKey":"11111",
* "value":[
* {
* "columnFamily":"first",
* "columnValue":{
* "name":"aaron",
* "age":18
* }
* },
* {
* "columnFamily":"second",
* "columnValue":{
* "name":"aaron",
* "age":18
* }
* }
* ]
* },
* {
* "rowKey":"22222",
* "value":[
* {
* "columnFamily":"first",
* "columnValue":{
* "name":"wwww",
* "age":19
* }
* },
* {
* "columnFamily":"second",
* "columnValue":{
* "name":"cccc",
* "age":18,
* "address","NanJ"
* }
* }
* ]
* }
* ]
* @throws IOException
* @author aaronChen
* @description 扫描全表数据
* @deprecated
*/
private static void scanTable(Connection connection, String tableName, String leftScanRowKey, String rightScanRowKey) throws IOException {
// 1.获取表对象
Table table = connection.getTable(TableName.valueOf(tableName));
// 2.构建Scan对象 // 左闭右开
Scan scan = new Scan(Bytes.toBytes(leftScanRowKey), Bytes.toBytes(rightScanRowKey));
// 3.扫描表
ResultScanner resultScanner = table.getScanner(scan);
// 4.解析resultScanner
for (Result result : resultScanner) {
// 5.解析result并打印
for (Cell cell : result.rawCells()) {
// 6.打印数据
System.out.println("RowKey:" + Bytes.toString(CellUtil.cloneRow(cell)) +
"，CF:" + Bytes.toString(CellUtil.cloneFamily(cell)) +
"，CN:" + Bytes.toString(CellUtil.cloneQualifier(cell)) +
"，Value:" + Bytes.toString(CellUtil.cloneValue(cell)));
}
}
// 7.关闭表连接
table.close();
}
/**
* @param tableName 要删除的表名
* @param rowKey 要删除的rowKey
* @param cf 要删除的列簇
* @param cn 要删除的列名
* @throws IOException
* @description Delete标记: 删除特定列列指定的版本
* DeleteFamily标记: 删除特定列族所有列
* DeleteColumn标记: 删除特定列的所有版本
* 指定rowkey: 使用DeleteFamily标记
* ---->不加时间戳表示删除[指定rowkey]的所有数据,加时间戳表示删除[指定rowkey]中[时间戳版本小于或等于指定时间戳]的所有数据
* 指定rowkey+columnFamily: 使用DeleteFamily标记
* ---->不加时间戳表示删除[指定列族]的所有数据,加了时间戳就表示删除[指定列族]下[时间戳版本小于或等于指定时间戳]的所有数据
* 指定rowkey+columnFamily+column(addColumns): 使用DeleteColumn标记
* ---->不加时间戳表示删除[指定列]所有版本的数据,加时间戳表示删除[指定列]中[时间戳版本小于或等于指定时间戳]的所有数据。
* 指定rowkey+columnFamily+column(addColumn): 使用Delete标记 (只删除单个版本数据,生产环境尽量别用)
* ---->不加时间戳表示删除[指定列]中[最新版本]的数据,加时间戳表示删除[指定列]中[指定时间戳版本]的数据。
* ---->不推荐的原因是:操作不同(如flush前后操作产生的结果会不一样)结果可能不同
* 如:在flush前如果有多个版本的数据,此时进行addColumn(不加时间戳)操作,会将最新版本的数据删除,然后老版本的数据会出现
* 在flush后进行addColumn(不加时间戳)操作,会将最新版本的数据删除,而此时flush已将老版本的数据进行了删除,所有此时老版本的数据就不会出现了
* <p>
* 删除列最好使用addColumns
* addColumns:不加时间戳表示删除指定列所有版本的数据(推荐)
* addColumns:加时间戳表示删除时间戳小于或等于指定时间戳的指定列的所有版本。
* addColumn:不加时间戳表示删除最新版本的数据,操作不同(如flush前后操作产生的结果会不一样)结果可能不同
* addColumn:加时间戳表示删除指定时间戳的指定列版本的数据。
*/
public static void deleteData(Connection connection, String tableName, String rowKey, String cf, String cn) throws IOException {
// 1.获取表对象
Table table = connection.getTable(TableName.valueOf(tableName));
// 2.创建删除对象
Delete delete = new Delete(Bytes.toBytes(rowKey));
//delete.addColumns(Bytes.toBytes(cf), Bytes.toBytes(cn));
//delete.addColumns(Bytes.toBytes(cf), Bytes.toBytes(cn),1574158036021L);
//delete.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn));
//delete.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn),1574158036021L);
// 2.2 删除指定的列族
// addFamily:删除指定列族的所有列的所有版本数据。
delete.addFamily(Bytes.toBytes(cf));
// 3.执行删除操作
table.delete(delete);
// 4.关闭连接
table.close();
}
/**
* @author aaronChen
* @description 关闭资源
*/
public static void close(Admin admin, Connection connection) {
if (admin != null) {
try {
admin.close();
} catch (IOException e) {
e.printStackTrace();
}
}
if (connection != null) {
try {
connection.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
public static JSONObject getJSONObject(String rowKey, List<Cell> cells) {
/**
* @return JSONObject {
* "rowKey":"11111",
* "value":{
* "name":"aaron",
* "age":18
* }
* }
*/
JSONObject jsonObject = new JSONObject();
JSONObject valueJsonObject = new JSONObject();
for (Cell cell : cells) {
valueJsonObject.put(Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toString(CellUtil.cloneValue(cell)));
}
jsonObject.put("rowKey", rowKey);
jsonObject.put("value", valueJsonObject);
return jsonObject;
}
}
比较核心的一个几个方法做出详细说明下：
3.2.1、新建表函数
public static boolean createTable(Admin admin, String tableName, boolean createNameSpaceIfExists, String... cfs)
会根据传入的相关信息进行表的创建，例如：
表名是"aaron1:test1"，当"aaron1"命名空间不存在的时候，会根据createNameSpaceIfExists参数来选择是否在不存在的情况下创建，如果需要创建，就会先创建该命名空间后，再去新建表。
public static void main(String[] args) throws Exception{
createTable(getHBaseConnection().getAdmin(),"aaron1:test1",false,"name","age","id");
}
当传入的createNameSpaceIfExists是true的时候
public static void main(String[] args) throws Exception{
createTable(getHBaseConnection().getAdmin(),"aaron1:test1",true,"name","age","id");
}
在web页面可以查看到已经创建成功
??
3.2.2、插入数据
?根据传入的参数来插入具体的数据。
public static void putData(Connection connection, String tableName, String rowKey, String columnFamily, String columnName, String value)
public static void main(String[] args) throws Exception{
putData(getHBaseConnection(),"aaron1:test1","aaaa","name","schoolName","Nanj");
}通过图形化工具可以看到数据已经插入成功
3.2.3、查询数据
本函数可以通过特定的rowKey查询出结果，并把结果转成JavaBean输出
public static <T> T getData(Connection connection, String tableName, String rowKey, String columnFamily, String columnName, boolean underScoreToCamel, Class<T> clz)
public static void main(String[] args) throws Exception {
PassengerInfor passengerInfor = getData(getHBaseConnection(), "aaron:test1", "11111", "first", "", true, PassengerInfor.class);
log.info("查询出来的结果为：{},", passengerInfor);
}
4、Flink异步IO访问Hbase
因为异步IO都是通过函数继承RichAsyncFunction这个抽象类，所以为了更大化的使用，先编写一个工具类。
4.1、异步IO访问Hbase工具类
/** * @param <IN> 流输入需要查询的内容 * @param <HBase> 从HBase查询返回的结果集 * @param <OUT> 最终输入流查询完HBase后返回的结果集 * @author ：aaronChen * @date ：Created in 2022-02-22 16:48 * @description： 该类是用来根据传入的一个rowKey来查询HBase某个表里的数据公共类 */
源码如下，其中使用的时候，使用到了如上的一些Hbase工具类：
package com.horse.utils.hbase;
import com.horse.utils.ThreadPoolUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import java.io.IOException;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;
/**
* @param <IN> 流输入需要查询的内容
* @param <HBase> 从HBase查询返回的结果集
* @param <OUT> 最终输入流查询完HBase后返回的结果集
* @author ：aaronChen
* @date ：Created in 2022-02-22 16:48
* @description： 该类是用来根据传入的一个rowKey来查询HBase某个表里的数据公共类
*/
@Slf4j
public abstract class HBaseAsyncFunctionUtil<IN, HBase, OUT> extends RichAsyncFunction<IN, OUT> {
private Connection connection;
private ThreadPoolExecutor threadPoolExecutor;
private Admin hBaseAdmin;
private HBase queryHBaseBean;
private Class<HBase> clz;
/**
* hBase表名
*/
private Table hBaseTable;
/**
* 列簇名
*/
private String columnFamily;
/**
* 列名
*/
private String columnName;
/**
* 命名空间:tableName
*/
private String tableName;
public HBaseAsyncFunctionUtil(String tableName, String columnFamily, String columnName, Class<HBase> clz) {
this(tableName, columnFamily, clz);
this.columnName = columnName;
}
public HBaseAsyncFunctionUtil(String tableName, String columnFamily, Class<HBase> clz) {
this.clz = clz;
this.tableName = tableName;
this.columnFamily = columnFamily;
}
public abstract String setRowKey(IN input);
public abstract OUT getList(String rowKey, IN input, HBase hBaseResult);
private void setQueryHBaseBean() throws Exception {
queryHBaseBean = clz.newInstance();
}
@Override
public void asyncInvoke(IN input, ResultFuture<OUT> resultFuture) throws Exception {
threadPoolExecutor.submit(new Runnable() {
@Override
public void run() {
resultFuture.complete(Collections.singleton(getResult(input)));
}
});
}
@Override
public void timeout(IN input, ResultFuture<OUT> resultFuture) throws Exception {
close();
connc();
log.info("Connection to HBase TimeOut,Now is reConnect!");
threadPoolExecutor.submit(new Runnable() {
@Override
public void run() {
resultFuture.complete(Collections.singleton(getResult(input)));
}
});
}
private OUT getResult(IN input) {
String rowKey = setRowKey(input);
try {
setQueryHBaseBean();
HBase hBaseResult = (HBase) HBaseUtils.getData(connection, tableName, rowKey, columnFamily, columnName, true, queryHBaseBean.getClass());
return getList(rowKey, input, hBaseResult);
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
//赋值connection
connc();
}
@Override
public void close() throws Exception {
super.close();
if (hBaseTable != null) {
hBaseTable.close();
}
if (hBaseAdmin != null) {
hBaseAdmin.close();
}
if (connection != null) {
connection.close();
}
}
private void connc() throws IOException {
connection = HBaseUtils.getHBaseConnection();
hBaseAdmin = connection.getAdmin();
if (!hBaseAdmin.tableExists(TableName.valueOf(tableName))) {
throw new IOException("Query Hbase Table is not exists!");
}
hBaseTable = connection.getTable(TableName.valueOf(tableName));
//初始化
threadPoolExecutor = ThreadPoolUtil.getThreadPool();
}
}
4.2、具体使用
同样还是利用nc模拟输入流数据
主类AsyncIOQueryHBase代码如下：
package com.horse.asyncio;
import com.horse.bean.PassengerInfor;
import com.horse.bean.UserLoginLog;
import com.horse.cep.function.MyFlatMapFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.horse.utils.hbase.HBaseAsyncFunctionUtil;
import java.util.concurrent.TimeUnit;
/**
* @author ：aaronChen
* @date ：Created in 2022-02-24 14:08
* @description ：
* @modifiedBy ：
*/
public class AsyncIOQueryHBase {
public static void main(String[] args) {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 8888);
SingleOutputStreamOperator<UserLoginLog> dataStream = socketTextStream.flatMap(new MyFlatMapFunction());
SingleOutputStreamOperator<PassengerInfor> outputStreamOperator = AsyncDataStream.unorderedWait(dataStream,
new HBaseAsyncFunctionUtil<UserLoginLog, PassengerInfor, PassengerInfor>("aaron:test1", "first", "", PassengerInfor.class) {
@Override
public String setRowKey(UserLoginLog input) {
return input.getUserName();
}
@Override
public PassengerInfor getList(String rowKey, UserLoginLog input, PassengerInfor hBaseResult) {
return hBaseResult;
}
}, 1000, TimeUnit.MINUTES, 100);
outputStreamOperator.print("异步IO查询Hbase结果输出");
try {
env.execute();
} catch (Exception e) {
e.printStackTrace();
}
}
}
在nc中输入如下数据：
{"loginId":11111,"loginTime":1645177352000,"loginStatus":1,"userName":"11111"}
查看日志：
?可以看到如上图中所示，已经能够从hbase中查询出结果。
如果觉得写的不错的，可以适当表示下哈～本人表示感谢。如果有写的不到位的也可以提问。
