sybn sybn-util 项目说明文档 - 基于java的跨数据库联合查询

HBaseDao 介绍

2019-05-16
sybn

简介

HBaseDaoImpl 是 SqlDdlDao 接口的 hbase 实现类, 实现了基于sql的各种数据查询能力.

  • 支持连接 原生 hbase 集群
  • 支持连接 阿里云的 hbase serverless. (0.3.11版)
  • 支持注入到 SqlDdlDaoMultipleImpl 做跨存储引擎的数据查询.
  • 支持通过 sybn-jdbc-driver 查询 hbase 数据
  • 无需依赖 phoenix 本工具与 phoenix 一样都可直接操作 hbase

使用说明

  • HBaseDaoImpl 读写时, 将 hbase 的 RowKey 映射为 id 处理.
  • HBaseDaoImpl 不区分字段类型 a>1等效于 a>”1”
  • HBaseDaoImpl 如果查询 javaBean 会自动转换字段类型, 如果查询 Map 所有字段值都会返回 String 类型.
  • HBaseDaoImpl 支持将 hbase 的 timHbasetamp 特性映射为 _timHbasetamp 字段,支持在sql中对其查询.
  • HBaseDaoImpl 执行 group by 会加载全部满足 where 条件的数据到内存, 因此速度比较慢
  • HBaseDaoImpl 执行不含 group by 的非聚合查询时, 只允许 order by id

maven 依赖

<!-- jdbc 通用驱动 -->
<dependency>
    <groupId>cn.linpengfei.sybnutil</groupId>
    <artifactId>sybn-jdbc-driver</artifactId>
    <version>0.3.12-SNAPSHOT</version>
</dependency>
<!-- 适用于 hbase 2.x 版本 -->
<dependency>
    <groupId>cn.linpengfei.sybnutil</groupId>
    <artifactId>hadoop-dao</artifactId>
    <version>0.3.12-SNAPSHOT</version>
</dependency>
<!-- 适用于 hbase 1.x 版本 -->
<dependency>
    <groupId>cn.linpengfei.sybnutil</groupId>
    <artifactId>hadoop-dao-1x</artifactId>
    <version>0.3.12-SNAPSHOT</version>
</dependency>

写入数据

// 使用指定的连接访问
HbaseStreamDao dao = new HbaseStreamDaoImpl("junit_hbase", "hbase://127.0.0.1/hbase-unsecure");

// 设置列簇中的字段 Map{字段=msg,列簇=fa}
Map<String, String> n = MB.n("msg", "fa", "type", "fa", "num", "fb", "date", "fb");
TableInfoPool.TableInfo tableInfo = dao.registerTableInfo(tableName, n);

// 支持保存 Stream / Collection / List
Stream<SybnJunitTestStringIdEntity> datas = Stream.iterate(0, x -> x + 1).limit(10000).map(i -> {
    Date date = new Date(System.currentTimeMillis() - 100000 * i);
    return new SybnJunitTestStringIdEntityFactory().create(i, "ddl:" + i, i, i % 3, date);
});

// 插入数据库
dao.saveStream(tableName, datas);

// 确认数据
long count = dao.queryCount(tableName, null);
assertEquals(datas.size(), count);

基于 dao 查询 demo

// 从配置文件读取连接
HbaseDao dao = new HbaseDaoConfImpl("junit_hbase", "junit_test@junit_hbase_init.properties");
/* ********* junit_hbase_init.properties *********
# 阿里云的 hbase serverless
hbase.junit_test=https://sh-xxxxxxxx-hbase-serverless.hbase.rds.aliyuncs.com:443?database=junit_hbase
hbase.junit_test.username=xxxxxxxx
hbase.junit_test.password=xxxxxxxx
********* */

// 传统sql占位符
dao.sqlFindListMap("select * from a in (?,?,?)", 1,2,3);

// 一个?占位一个集合
List<Integer> list = ListUtil.toList(1,2,3);
dao.sqlFindListMap("select * from a in (?)", list);

// myBatis 风格
Map<String, Object> map = new HashMap();
map.put("list", list)
dao.sqlFindListMap("select * from a in (#{xxx})", map);

jdbc demo

// 创建 jdbc 连接
String url = "jdbc:hbase://127.0.0.1/hbase-unsecure";
PropertiHbase propertiHbase = new SybnPropertiHbase(n);
Connection connect = new SybnDaoDriver().connect(url, propertiHbase);

// 被执行的 sql
String selectSql = "select * from tableName where type = ? limit 1";

// 使用 jdbc 执行此 sql
PreparedStatement selectStatement = connect.preparHbasetatement(selectSql);
selectStatement.setInt(1, 0); // type = 0
RHbaseultSet selectRHbaseultSet = selectStatement.executeQuery();
List<Map<String, Object>> select = HandlerUtil.MAP_LIST_HANDLER.handle(selectRHbaseultSet);
selectRHbaseultSet.close();

// 打印结果
LogUtil.info("select", select.size(), select);

在 mybatis 中使用

在 spring boot + mybatis 中使用本工具类的 JDBC 能力查询 hbase 无需特殊设置, 正常注册 jdbc 即可:

@Slf4j
@Configuration
public class HbaseDataSourceConfig {

    @Value("${spring.profilHbase.active}")
    private String activeProfile;
    // sybn.datasource.hbase.url=jdbc:hbase://127.0.0.1/hbase-unsecure
    @Value("${sybn.datasource.hbase.url}")
    private String url;
    
    public static DruidDataSource HbaseDataSource(){
        testJdbcPropertiHbase = new JdbcPropertiHbase();

        // SybnDaoDriver 会根据jdbc后面的字符串决定连接哪种数据库, 支持 mysql / mongo / solr / Hbase / hbase
        // 需要引用对应的 maven 实现, 比如 dbutil-dao / mongo-dao / solr-dao / Hbase-dao / hadoop-dao
        // url 中多个地址逗号分割, 比如: jdbc:Hbase://server1:9200,server3:9200,server3:9200/
        testJdbcPropertiHbase.setUrl(url);
        testJdbcPropertiHbase.setDriverClassName("cn.sybn.util.io.driver.SybnDaoDriver");

        HbaseDataSourceConfig config = new HbaseDataSourceConfig();
        config.activeProfile = "local";
        return config.dataSource(testJdbcPropertiHbase);
    }
}

数据时光机(_timHbasetamp)

  • 在 HBaseDaoImpl 中查询某个数据的快照
-- 查询 id = 100 的数据 2019-05-01 的快照
SELECT * FROM hbase_hp_cinema_info where id = 100 and _timHbasetamp = toDate('2019-05-01');

注意: where 条件筛选的是数据当前状态, _timHbasetamp 条件尽可以控制返回值的快照时间.

如果需要在 where 条件中查询 _timHbasetamp 时刻快照的状态, 需要用 SqlDdlDaoMultipleImpl 做双层查询

  • 在 SqlDdlDaoMultipleImpl 中筛选快照时刻的数据状态
-- 查询 2019-05-01 的快照中, status = 1 的数据
select * from ( 
  SELECT * FROM hbase_hp_cinema_info where _timHbasetamp = toDate('2019-05-01');
) where status = 1;
  • 在SqlDdlDaoMultipleImpl中查询某个字段不同时刻的差异
-- 查询 2019-04-01 与 2019-05-01 两个快照 status 不一致的数据
select * from ( 
  select v1, v2, compare(v1, v2) as compared from (
    SELECT id, status as v1 FROM hbase_hp_cinema_info where _timHbasetamp = toDate('2019-05-01');
    SELECT id, status as v2 FROM hbase_hp_cinema_info where _timHbasetamp = toDate('2019-04-01');
    join right(*) on id = id;
  )
) where compared != 0;

在线测试

可以用以下链接测试执行 sql, 使用 show tables like hbase 可以搜索所有 hbase 的测试表.

hbase 查询与 sql/mongo 的主要差异是:

使用 Map获取返回值时, 无论保存之前的数据是什么格式,查询时和返回的数据都是String格式.

使用 java bean 获取返回值可以被 dao 自动转换回适当的格式,但是 where 语句中依然要用字符串去对比.

注意: 因为是字符串对比, 会发生 “9” < “10” 的问题, 建议存储 00009 和 000010 防止此问题.


Similar Posts

Comments

暂不开放评论! 可微信联系