Hbase二级索引_Hive on Hbase 及phoenix详解
文章目录
[*]Hbase二级索引_Hive on Hbase 及phoenix详解
[*]
[*]知识点02：课程目标
[*]知识点03：SQL on Hbase
[*]知识点04：Hive on Hbase 介绍
[*]知识点05：Hive on Hbase 配置
[*]知识点06：Hive on Hbase 实现
[*]知识点07：二级索引问题
[*]知识点08：Phoenix的介绍
[*]知识点09：Phoenix的安装配置
[*]知识点10：Phoenix的语法：DDL：NS
[*]知识点11：Phoenix的语法：DDL：Table
[*]知识点12：Phoenix的语法：DML：upsert
[*]知识点13：Phoenix的语法：DML：delete
[*]知识点14：Phoenix的语法：DQL：select
[*]知识点15：Phoenix的使用：预分区
[*]知识点16：Phoenix的使用：加盐salt
[*]知识点17：Phoenix的使用：视图
[*]知识点18：Phoenix的使用：JDBC
Hbase二级索引_Hive on Hbase 及phoenix详解
知识点02：课程目标
[*]SQL on Hbase
[*]使用SQL语句来操作Hbase
[*]Hbase不支持SQL接口
[*]额外的工具来实现
[*]Hive on Hbase【了解】
[*]使用Hive中的SQL语句来实现对Hbase数据的操作
[*]本质：通过MapReduce来实现读写Hbase
[*]Phoenix【重点】
[*]专门为Hbase所设计的一个工具
[*]本质：直接封装Hbase的JavaAPI来实现的
[*]功能、应用场景、基本原理、特点
[*]基本使用：语法【upsert、delete、select】
知识点03：SQL on Hbase
[*] 问题
[*]Hbase是列存储NoSQL，不支持SQL，开发接口不方便大部分用户使用，怎么办？
[*] 分析
[*] 应用场景：应用系统或者大数据存储系统
[*] 大数据存储系统：大数据工程师
[*]利用Hbase来存储大量要分析处理的数据
[*]使用JavaAPI通过MapReduce或者通过Spark来实现数据的读写
[*]Java
[*]Scala
[*] 应用系统：Java工程师、数据分析师
[*]利用Hbase来存储大量的商品数据、订单数据，来提供高性能的查询
[*] 问题：Java人员不会Hbase Java API，对于数据库会JDBC
[*] 解决：需要一个工具能让Hbase支持SQL，支持JDBC方式对Hbase进行处理
[*] Hbase的结构是否能实现基于SQL的查询操作？
[*] 普通表数据：按行操作
idnameagesexaddr
001zhangsan18nullshanghai
002lisi20femalenull
003wangwunullmalebeijing
……
[*] Hbase数据：按列操作
rowkeycf1:idcf1:namecf1:agecf2:sexcf2:addr
zhangsan_001001zhangsan18nullshanghai
lisi_002002lisi20femalenull
wangwu_003003wangwunullmalebeijing
……
[*] 可以基于Hbase数据构建结构化的数据形式
[*] 可以用SQL来实现处理
[*] 实现
[*]将Hbase表中每一行对应的所有列构建一张完整的结构化表
[*]如果这一行没有这一列，就补null
[*]Hive：通过MapReduce来实现
[*]Phoenix：通过Hbase API封装实现的
[*] 总结
[*]原因：满足各种应用场景下，对于Hbase使用的方式，基于SQL方式会更加通用
[*]实现：将整张表的数据构建结构化形式，每一行没有列就补null
[*]原理：将SQL转换成了Hbase的客户端操作来实现的
知识点04：Hive on Hbase 介绍
[*] 功能：实现Hive与Hbase集成，使用Hive SQL对Hbase的数据进行处理
[*] 原理
[*] Hive的功能：使用HQL对表的数据进行处理
[*] 本质：通过MapReduce对HDFS中的文件进行处理
[*] 原理
[*] TextInputFormat：读文件
[*] TextOutputFormat：写文件
[*] MapReduce的功能：读取数据进行分布式计算
[*] InputFormat：输入类
[*]TextInputFormat：默认的输入类，用于读取文件系统
[*]DBInputFormat：用于读取JDBC数据库
[*]实现Sqoop导入的：将MySQL数据导入到Hive或者HDFS
[*]TableInputFormat：用于读取Hbase数据
[*] OutputFormat：输出类
[*] TextOutputFormat：默认的输出类，用于将结果写入文件系统
[*] DBOutputFormat：用于写入JDBC数据库
[*]实现Sqoop导出的：将HDFS数据写入MySQL
[*] TableOutputFormat：用于写入HBase数据库
[*] 原理：Hive可以通过MapReduce来实现映射读写Hbase表的数据
[*] 特点
[*] 优点：支持完善的SQL语句，可以实现各种复杂SQL的数据处理及计算，通过分布式计算程序实现，对大数据量的数据处理比较友好
[*] 缺点：不支持二级索引，数据量不是特别大的情况下，性能一般
[*] 应用
[*]基于大数据高性能的离线读写，并且使用SQL来开发
知识点05：Hive on Hbase 配置
[*] 需求
[*]配置Hive与Hbase集成，实现Hive中可以读写Hbase表
[*] 分析
[*]step1：修改Hive配置文件，指定Hbase的Zookeeper地址
[*]step2：按顺序启动HDFS、ZK、Hbase、Hive
[*] 实现
[*] 全部操作在第三台机器
[*] 修改hive-site.xml：Hive通过SQL访问Hbase，就是Hbase的客户端，就要连接zookeeper
cd /export/server/hive-2.1.0-bin/
vim conf/hive-site.xml
<property>
<name>hive.zookeeper.quorum</name>
<value>node1,node2,node3</value>
</property>
<property>
<name>hbase.zookeeper.quorum</name>
<value>node1,node2,node3</value>
</property>
<property>
<name>hive.server2.enable.doAs</name>
<value>false</value>
</property>
[*] 修改hive-env.sh
export HBASE_HOME=/export/server/hbase-2.1.0
[*] 启动HDFS、ZK、Hbase：第一台机器
start-dfs.sh
/export/server/zookeeper-3.4.6/bin/start-zk-all.sh
start-hbase.sh
[*] 启动Hive和YARN：第三台机器
#启动YARN
start-yarn.sh
#先启动metastore服务
start-metastore.sh
#然后启动hiveserver
start-hiveserver2.sh
#然后启动beeline
start-beeline.sh
[*] 总结
[*]先配置Hive的配置文件：添加Hbase的地址
[*]然后按照先后顺序启动即可
知识点06：Hive on Hbase 实现
[*] 需求
[*]在Hive中实现对Hbase表的数据读写
[*] 分析
[*]step1：如果表在Hbase中没有，Hive中没有，在Hive中创建表，指定在Hbase中创建关联表
[*]场景比较少
[*]在Hive中建一张表，自动在Hbase中也创建一张对应的表
[*]step2：如果表在Hbase中有，但是Hive中没有，Hive中创建一张外部表，关联Hbase表
[*]主要应用的方式
[*]Hbase中的表已经存在，已经有数据，构建一张Hive关联表，使用SQL进行查询
[*] 实现
[*] 第三台机器测试
[*] 如果Hbase中表不存在：【用的比较少】
[*] 创建测试数据文件
vim /export/data/hive-hbase.txt
1,zhangsan,80
2,lisi,60
3,wangwu,30
4,zhaoliu,70
[*] 创建测试表
--创建测试数据库
create database course;
--切换数据库
use course;
--创建原始数据表
create external table if not exists course.score(
id int,
cname string,
score int
) row format delimited fields terminated by ',' stored as textfile ;
--加载数据文件
load data local inpath '/export/data/hive-hbase.txt' into table score;
[*] 创建一张Hive与HBASE的映射表
create table course.hbase_score(
id int,
cname string,
score int
)
stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
with serdeproperties("hbase.columns.mapping" = "cf:name,cf:score")
tblproperties("hbase.table.name" = "hbase_score");
[*] 将测试表的数据写入映射表
set hive.exec.mode.local.auto=true;
insert overwrite table course.hbase_score select id,cname,score from course.score;
[*] 如果Hbase中表已存在，只能创建外部表
create external table course.t1(
key string,
name string,
agestring,
addr string,
phone string
)
stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
with serdeproperties("hbase.columns.mapping" = ":key,basic:name,basic:age,other:addr,other:phone")
tblproperties("hbase.table.name" = "itcast:t1");
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-gNcVCpdb-1616545523608)(20210323_分布式NoSQL列存储数据库Hbase（六）.assets/image-20210323160741636.png)]
[*] 总结
[*]Hive中的只是关联表，并没有数据，数据存储在Hbase表中
[*] 在Hive中创建Hbase的关联表，关联成功后，使用SQL处理关联表
[*]如果Hbase中表不存在，默认使用Hive的第一列作为rowkey
[*]如果Hbase中表已存在，只能建外部表，使用:key来表示rowkey
[*]HIve中与Hbase关联的表，不能使用load加载，只能使用insert，通过MR读写数据
知识点07：二级索引问题
[*] 问题
[*]Hbase使用Rowkey作为唯一索引，需要构建二级索引来解决查询问题，如何构建二级索引以及维护索引表？
[*] 分析
[*] step1：基于存储和常用查询需求，构建数据表
[*] step2：基于其他查询需求，构建索引表
[*] step3：先查询索引表，再查询数据表
[*] step4：自动维护索引表与原始数据表的数据一致性
[*] 实现
[*] 构建数据表
rowkey:name_ididnameagesexaddr
zhangsan_001001zhangsan18maleshanghai
lisi_002002lisi18femalebeijing
zhangsan_003003zhangsan20male
……
[*] 构建索引表
rowkey:id_namecol：原始数据表的rowkey
001_zhangsanzhangsan_001
002_lisilisi_002
003_zhangsanzhangsan_003
……
[*] 查询：根据id查询
[*]先查询索引表，获取原表的Rowkey
[*]再根据原表Rowkey查询原表的数据
[*] 维护
[*]当原表数据需要进行增删改时，索引表自动进行同步增删改对应的数据，保持一致性
[*] 解决方案
[*] 方案一：客户端操作实现
put1
put2
table1.put(put1)
table2.put(put2)
[*] 方案二：协处理器实现
[*]自己开发代码
[*]让Hbase监听原表，原表更改一条，Hbase自动对索引表更改一条
[*]缺点：开发比较麻烦
[*] 方案三：第三方工具
[*] Phoenix：将所有协处理器都封装好了
[*] 支持SQL
[*] 支持自动二级索引的构建及维护
create index
[*] 总结
[*]需求：必须根据不同的查询条件，创建不同的索引表，并且维护所有索引表与原始数据表的同步
[*]解决：通过Phoenix自带的协处理器来实现
知识点08：Phoenix的介绍
[*] 功能
[*] 专门基于Hbase所设计的SQL on Hbase 工具
[*] 使用Phoenix实现基于SQL操作Hbase
[*] 使用Phoenix自动构建二级索引并维护二级索引
[*] 原理
[*] 上层提供了SQL接口
[*]底层全部通过Hbase Java API来实现，通过构建一系列的Scan和Put来实现数据的读写
[*] 功能非常丰富
[*]底层封装了大量的内置的协处理器，可以实现各种复杂的处理需求，例如二级索引等
[*] 特点
[*]优点
[*]支持SQL接口
[*]支持自动维护二级索引
[*]缺点
[*]SQL支持的语法不全面
[*]Bug比较多
[*]Hive on Hbase对比
[*]Hive：SQL更加全面，但是不支持二级索引，底层通过分布式计算工具来实现
[*]Phoenix：SQL相对支持不全面，但是性能比较好，直接使用HbaseAPI，支持索引实现
[*] 应用
[*] Phoenix适用于任何需要使用SQL或者JDBC来快速的读写Hbase的场景
[*] 或者需要构建及维护二级索引场景
知识点09：Phoenix的安装配置
[*] 需求
[*]http://phoenix.apache.org/
[*]安装部署配置Phoenix，集成Hbase
[*] 分析
[*]step1：上传解压安装
[*]step2：修改配置，指定Hbase连接地址
[*]step3：启动Phoenix，连接Hbase
[*] 实现
[*] 下载：http://phoenix.apache.org/download.html
[*] 第一台机器上传
cd /export/software/
rz
[*] 第一台机器解压
tar -zxvf apache-phoenix-5.0.0-HBase-2.0-bin.tar.gz -C /export/server/
cd /export/server/
mv apache-phoenix-5.0.0-HBase-2.0-bin phoenix-5.0.0-HBase-2.0-bin
[*] 修改三台Linux文件句柄数
vim /etc/security/limits.conf
#在文件的末尾添加以下内容，*号不能去掉
* soft nofile 65536
* hard nofile 131072
* soft nproc 2048
* hard nproc 4096
[*] 将Phoenix所有jar包分发到Hbase的lib目录下
#拷贝到第一台机器
cd /export/server/phoenix-5.0.0-HBase-2.0-bin/
cp phoenix-* /export/server/hbase-2.1.0/lib/
cd /export/server/hbase-2.1.0/lib/
#分发给第二台和第三台
scp phoenix-* node2:$PWD
scp phoenix-* node3:$PWD
[*] 修改hbase-site.xml，添加一下属性
cd /export/server/hbase-2.1.0/conf/
vim hbase-site.xml
<!-- 关闭流检查，从2.x开始使用async -->
<property>
<name>hbase.unsafe.stream.capability.enforce</name>
<value>false</value>
</property>
<!-- 支持HBase命名空间映射 -->
<property>
<name>phoenix.schema.isNamespaceMappingEnabled</name>
<value>true</value>
</property>
<!-- 支持索引预写日志编码 -->
<property>
<name>hbase.regionserver.wal.codec</name>
<value>org.apache.hadoop.hbase.regionserver.wal.IndexedWALEditCodec</value>
</property>
<!-- 配置NS映射 -->
<property>
<name>phoenix.schema.isNamespaceMappingEnabled</name>
<value>true</value>
</property>
[*] 同步给其他两台机器
scp hbase-site.xml node2:$PWD
scp hbase-site.xml node3:$PWD
[*] 同步给Phoenix
cp hbase-site.xml /export/server/phoenix-5.0.0-HBase-2.0-bin/bin/
[*] 重启Hbase
stop-hbase.sh
start-hbase.sh
[*] 启动Phoenix
cd /export/server/phoenix-5.0.0-HBase-2.0-bin/
bin/sqlline.py node1:2181
[*] 测试
!tables
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KuMfWPJo-1616545523612)(20210323_分布式NoSQL列存储数据库Hbase（六）.assets/image-20210323170434725.png)]
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-nqVKRbLI-1616545523613)(20210323_分布式NoSQL列存储数据库Hbase（六）.assets/image-20210323170543556.png)]
[*] 总结
[*]解压安装
[*]修改配置
[*]启动服务
[*]测试环境
知识点10：Phoenix的语法：DDL：NS
[*] http://phoenix.apache.org/language/index.html
[*] 需求
[*]实现基于SQL的数据库管理：创建、切换、删除
[*] 分析
[*]step1：创建Namespace
[*]step2：切换Namespace
[*]step3：删除Namespace
[*] 实现
[*] 创建NS
create schema if not exists student;
[*] 切换NS
use student;
[*] 删除NS
drop schema if exists student;
[*] 总结
[*]基本与SQL语法一致
[*]注意：Phoenix中默认会将所有字符转换为大写，如果想要使用小写字母，必须加上双引号
知识点11：Phoenix的语法：DDL：Table
[*] 需求
[*]实现基于SQL的数据表管理：创建、列举、查看、删除
[*] 分析
[*]step1：列举当前所有的表
[*]step2：创建表
[*]step3：查询表信息
[*]step4：删除表
[*] 实现
[*] 列举
!tables
[*] 创建
[*] 语法：http://phoenix.apache.org/language/index.html#create_table
CREATE TABLE my_schema.my_table (
id BIGINT not null primary key,
date Date
);
CREATE TABLE my_table (
id INTEGER not null primary key desc,
m.date DATE not null,
m.db_utilization DECIMAL,
i.db_utilization
) m.VERSIONS='3';
CREATE TABLE stats.prod_metrics (
host char(50) not null,
created_date date not null,
txn_count bigint
CONSTRAINT pk PRIMARY KEY (host, created_date)
);
CREATE TABLE IF NOT EXISTS "my_case_sensitive_table"(
"id" char(10) not null primary key,
"value" integer
) DATA_BLOCK_ENCODING='NONE',VERSIONS=5,MAX_FILESIZE=2000000
split on (?, ?, ?);
CREATE TABLE IF NOT EXISTS my_schema.my_table (
org_id CHAR(15),
entity_id CHAR(15),
payload binary(1000),
CONSTRAINT pk PRIMARY KEY (org_id, entity_id)
) TTL=86400
[*] 如果Hbase中没有这个表
use default;
create table if not exists ORDER_DTL(
ID varchar primary key,
C1.STATUS varchar,
C1.PAY_MONEY float,
C1.PAYWAY integer,
C1.USER_ID varchar,
C1.OPERATION_DATE varchar,
C1.CATEGORY varchar
);
[*] 如果Hbase中已存在会自动关联
create table if not exists ORDER_INFO(
"ROW" varchar primary key,
"C1"."USER_ID" varchar,
"C1"."OPERATION_DATE" varchar,
"C1"."PAYWAY" varchar,
"C1"."PAY_MONEY" varchar,
"C1"."STATUS" varchar,
"C1"."CATEGORY" varchar
) column_encoded_bytes=0 ;
[*] 查看
!desc order_info;
[*] 删除
drop table if exists order_dtl;
[*] 总结
[*] 创建表时，必须指定主键作为Rowkey，主键列不能加列族
create table if not exists ORDER_INFO(
–不能这么写
“C1”.“ROW” varchar primary key,
“C1”.“USER_ID” varchar,
“C1”.“OPERATION_DATE” varchar,
“C1”.“PAYWAY” varchar,
“C1”.“PAY_MONEY” varchar,
“C1”.“STATUS” varchar,
“C1”.“CATEGORY” varchar
) column_encoded_bytes=0 ;
- Phoenix 4.8版本之前只要创建同名的Hbase表，会自动关联数据
- Phoenix 4.8版本以后，不推荐关联表的方式
- 推荐使用视图关联的方式来实现，如果你要使用关联表的方式，必须加上以下参数
```
column_encoded_bytes=0 ;
```
- 如果关联已存在的表，Rowkey字段叫做ROW，使用时必须加上双引号
select “ROW”,“C1”.USER_ID,“C1”.“PAYWAY” from ORDER_INFO;
知识点12：Phoenix的语法：DML：upsert
列名数值描述Rowkey02602f66-adc7-40d4-8485-76b5632b5b53行健，编码生成USER_ID4944191用户idOPERATION_DATE2020-04-25 12:09:16操作时间PAYWAY1支付方式PAY_MONEY4070支付金额STATUS已提交提交状态CATEGORY手机;分类
[*] 需求
[*]基于order_info订单数据实现DML插入数据
[*] 分析
[*] Phoenix中插入更新的命令为：upsert
[*]功能：insert + update
[*]MySQL：replace
[*]如果存在就更新，如果不存在就插入
[*] 语法及示例
UPSERT INTO TEST VALUES('foo','bar',3);
UPSERT INTO TEST(NAME,ID) VALUES('foo',123);
UPSERT INTO TEST(ID, COUNTER) VALUES(123, 0) ON DUPLICATE KEY UPDATE COUNTER = COUNTER + 1;
UPSERT INTO TEST(ID, MY_COL) VALUES(123, 0) ON DUPLICATE KEY IGNORE;
[*] 实现
[*] 插入一条数据
upsert into order_info values('z8f3ca6f-2f5c-44fd-9755-1792de183845','4944191','2020-04-25 12:09:16','1','4070','未提交','电脑');
[*] 更新USERID为123456
upsert into order_info("ROW","USER_ID") values('z8f3ca6f-2f5c-44fd-9755-1792de183845','123456');
[*] 总结
[*] 语法类似于insert语法
[*] 功能：insert + update
知识点13：Phoenix的语法：DML：delete
[*] 需求
[*]基于order_info订单数据实现DML删除数据
[*] 分析
[*] Phoenix中插入更新的命令为：delete
[*] 语法及示例
DELETE FROM TEST;
DELETE FROM TEST WHERE ID=123;
DELETE FROM TEST WHERE NAME LIKE 'foo%';
[*] 实现
[*] 删除USER_ID为123456的rowkey数据
delete from order_info where USER_ID = '123456';
[*] 总结
[*]与MySQL是一致的
知识点14：Phoenix的语法：DQL：select
[*] 需求
[*]基于order_info订单数据实现DQL查询数据
[*] 分析
[*] Phoenix中插入更新的命令为：select
[*] 语法及示例
SELECT * FROM TEST LIMIT 1000;
SELECT * FROM TEST LIMIT 1000 OFFSET 100;
SELECT full_name FROM SALES_PERSON WHERE ranking >= 5.0
UNION ALL SELECT reviewer_name FROM CUSTOMER_REVIEW WHERE score >= 8.0
[*] 实现
[*] 查询支付方式为1的数据
select "ROW",payway,pay_money,category from order_info where payway = '1';
[*] 查询每种支付方式对应的用户人数，并且按照用户人数降序排序
[*]分组：每、各个、不同
[*]排序：用户人数
select
payway,
count(distinct user_id) as numb
from order_info
group by payway
order by numb desc;
[*] 查询数据的第60行到66行
--以前的写法：limit M,N
--M：开始位置
--N：显示的条数
--Phoenix的写法：limit N offset M
select * from order_info limit 6 offset 60;//总共66行，显示最后6行
[*] 函数支持
[*]http://phoenix.apache.org/language/functions.html
[*] 总结
[*]基本查询与MySQL也是一致的
[*]写的时候注意数据类型以及大小写的问题即可
[*]如果遇到SQL报错，检查语法是否支持
知识点15：Phoenix的使用：预分区
知识点16：Phoenix的使用：加盐salt
知识点17：Phoenix的使用：视图
[*] 需求
[*]直接关联Hbase中的表，会导致误删除，对数据的权限会有影响，容易出现问题，如何避免？
[*] 分析
[*]Phoenix中建议使用视图的方式来关联Hbase中已有的表
[*]通过构建关联视图，可以解决大部分数据查询的数据，不影响数据
[*]视图：理解为只读的表
[*] 实现
[*] 创建视图，关联Hbase中已经存在的表
create view if not exists "MOMO_CHAT"."MSG" (
"pk" varchar primary key, -- 指定ROWKEY映射到主键
"C1"."msg_time" varchar,
"C1"."sender_nickyname" varchar,
"C1"."sender_account" varchar,
"C1"."sender_sex" varchar,
"C1"."sender_ip" varchar,
"C1"."sender_os" varchar,
"C1"."sender_phone_type" varchar,
"C1"."sender_network" varchar,
"C1"."sender_gps" varchar,
"C1"."receiver_nickyname" varchar,
"C1"."receiver_ip" varchar,
"C1"."receiver_account" varchar,
"C1"."receiver_os" varchar,
"C1"."receiver_phone_type" varchar,
"C1"."receiver_network" varchar,
"C1"."receiver_gps" varchar,
"C1"."receiver_sex" varchar,
"C1"."msg_type" varchar,
"C1"."distance" varchar
);
[*] 查询数据
select
"pk",
"C1"."msg_time",
"C1"."sender_account",
"C1"."receiver_account"
from "MOMO_CHAT"."MSG"
limit 10;
[*] 总结
[*]工作中主要构建的都是视图
[*]MySQL：视图
[*]Hive：外部表
[*]Phoenix：视图
知识点18：Phoenix的使用：JDBC
[*] 需求
[*]工作中实际使用SQL，会基于程序中使用JDBC的方式来提交SQL语句，在Phoenix中如何实现？
[*] 分析
[*] Phoenix支持使用JDBC的方式来提交SQL语句
[*] 例如：聊天分析案例中需求：查询条件为日期【年-月-日】 + 发送人ID + 接受人ID
select
*
from "MOMO_CHAT"."MSG"
where
substr("msg_time",0,10) = '2021-03-22'
and "sender_account" = '17351912952'
and "receiver_account" = '17742251415';
[*] 可以在代码中基于JDBC来提交SQL查询
[*] 实现
[*] 构建JDBC连接Phoenix
package cn.itcast.momo_chat.service.impl;
import cn.itcast.momo_chat.entity.Msg;
import cn.itcast.momo_chat.service.ChatMessageService;
import org.apache.phoenix.jdbc.PhoenixDriver;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;
/**
* @ClassName PhoenixChatMessageService
* @Description TODO JDBC连接Phoenix实现数据查询
* @Create By Frank
*/
public class PhoenixChatMessageService implements ChatMessageService {
private Connection connection;
public PhoenixChatMessageService() throws ClassNotFoundException, SQLException {
try {
//申明驱动类
Class.forName(PhoenixDriver.class.getName());
// System.out.println(PhoenixDriver.class.getName());
//构建连接
connection = DriverManager.getConnection("jdbc:phoenix:node1,node2,node3:2181");
} catch (ClassNotFoundException e) {
throw new RuntimeException("加载Phoenix驱动失败!");
} catch (SQLException e) {
throw new RuntimeException("获取Phoenix JDBC连接失败!");
}
}
@Override
public List<Msg> getMessage(String date, String sender, String receiver) throws Exception {
PreparedStatement ps = connection.prepareStatement(
"SELECT * FROM MOMO_CHAT.MSG T WHERE substr(\"msg_time\", 0, 10) = ? "
+ "AND T.\"sender_account\" = ? "
+ "AND T.\"receiver_account\" = ? ");
ps.setString(1, date);
ps.setString(2, sender);
ps.setString(3, receiver);
ResultSet rs = ps.executeQuery();
List<Msg> msgList = new ArrayList<>();
while(rs.next()) {
Msg msg = new Msg();
msg.setMsg_time(rs.getString("msg_time"));
msg.setSender_nickyname(rs.getString("sender_nickyname"));
msg.setSender_account(rs.getString("sender_account"));
msg.setSender_sex(rs.getString("sender_sex"));
msg.setSender_ip(rs.getString("sender_ip"));
msg.setSender_os(rs.getString("sender_os"));
msg.setSender_phone_type(rs.getString("sender_phone_type"));
msg.setSender_network(rs.getString("sender_network"));
msg.setSender_gps(rs.getString("sender_gps"));
msg.setReceiver_nickyname(rs.getString("receiver_nickyname"));
msg.setReceiver_ip(rs.getString("receiver_ip"));
msg.setReceiver_account(rs.getString("receiver_account"));
msg.setReceiver_os(rs.getString("receiver_os"));
msg.setReceiver_phone_type(rs.getString("receiver_phone_type"));
msg.setReceiver_network(rs.getString("receiver_network"));
msg.setReceiver_gps(rs.getString("receiver_gps"));
msg.setReceiver_sex(rs.getString("receiver_sex"));
msg.setMsg_type(rs.getString("msg_type"));
msg.setDistance(rs.getString("distance"));
msgList.add(msg);
}
return msgList;
}
@Override
public void close() {
try {
connection.close();
} catch (SQLException e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws Exception {
ChatMessageService chatMessageService = new PhoenixChatMessageService();
List<Msg> message = chatMessageService.getMessage("2021-03-22", "17351912952", "17742251415");
for (Msg msg : message) {
System.out.println(msg);
}
chatMessageService.close();
}
}
[*] 运行查看结果
[*] 总结
[*]Phoenix支持SQL
[*]支持JDBC方式提交SQL语句实现数据处理
页:
[1]