1、HBase的namespace介绍
namespace：名字空间或命名空间，相当于mysql中的database。
2、namespace的作用
1）配额管理：限制一个namespace可以使用的资源，包括region和table
2）命名空间安全管理：提供了另一个层面的多租户安全管理
3）Region服务器组：一个命名或一张表，可以被固定到一组RegionServers上，从而保证了数据隔离性
3、namespace的基本操作
1）列出所有的namespace
list_namespace
2）创建namespace
create_namespace 'nametest' 3）查看namespace
describe_namespace 'nametest'4）在namespace下创建表
create 'nametest:testtable', 'fm1' 5）查看namespace下的表
list_namespace_tables 'nametest' 6）删除namespace（删除空间的时候，命名空间下的表必须是空，否则先删除表）
drop_namespace 'nametest' 4、Hbase的数据版本的确界以及TTL（数据存活的时间）
hbase中有版本version 的概念
hbase中针对版本有一个上界和下届的概念
1）版本的下届：最小的版本数据，下届的默认值是0，说明最小版本数是1
2）版本的上界：数据最大的版本，上届的默认值是1。
TTL：数据的存活周期
可以设置在某一个列族上或者是设置在一条数据上（rowkey的数据）
5、通过代码实现版本设定以及数据的TTL
1）创建maven工程并添加jar包
2）代码实现
public class HBaseVersionAndTTL {
public static void main(String[] args) throws IOException, InterruptedException {
// 创建数据库的连接
Configuration configuration = HBaseConfiguration.create(); configuration.set("hbase.zookeeper.quorum","hadoop01:2181,hadoop02:2181,hadoop03:2181");
Connection connection = ConnectionFactory.createConnection();
Admin admin = connection.getAdmin();
// 创建一张表
if(!admin.tableExists(TableName.valueOf("version_hbase"))){
HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf("version_hbase"));
HColumnDescriptor f1 = new HColumnDescriptor("f1");
// 设置版本
f1.setMinVersions(3);
f1.setMaxVersions(5);
//针对某一个列族下面所有的列设置TTL
f1.setTimeToLive(30); // 存活周期
hTableDescriptor.addFamily(f1);
admin.createTable(hTableDescriptor);
}
Table version_hbase = connection.getTable(TableName.valueOf("version_hbase"));
Put put = new Put("1".getBytes());
//针对某一条具体的数据设置TTL
//put.setTTL(3000);
// 指定数据的插入时间 put.addColumn("f1".getBytes(),"name".getBytes(),System.currentTimeMillis(),"zhangsan".getBytes());
version_hbase.put(put);
Thread.sleep(1000);
Put put2 = new Put("1".getBytes());
put2.addColumn("f1".getBytes(),"name".getBytes(),System.currentTimeMillis(),"zhangsan2".getBytes());
version_hbase.put(put2);
Get get = new Get("1".getBytes());
get.setMaxVersions();
Result result = version_hbase.get(get);
Cell[] cells = result.rawCells();
for (Cell cell : cells) {
System.out.println(Bytes.toString(CellUtil.cloneValue(cell)));
}
version_hbase.close();
connection.close();
}
}
6、微博案列
1）三张表
2）创建maven工程导入包
3）拷贝配置文件daomaven下
/export/servers/hbase-1.2.0-cdh5.14.0/conf 这个路径下的三个配置文件，分别是 core-site.xml、hdfs-site.xml、hbase-site.xml三个配置文件，拷贝到maven工程的resources资源目录下
4）创建命名空间以及表名的定义
private String WEIBO_CONTENT = "weibo:content";
private String WEIBO_RELATIONS = "weibo:relations";
private String WEIBO_CONTENT_EMAILS = "weibo:receive_content_email";
private Connection connection;
private Admin admin;
@Before
public void init() throws IOException {
//构建config对象
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", "hadoop01:2181,hadoop02:2181");
//数据库的连接
connection = ConnectionFactory.createConnection(conf);
admin = connection.getAdmin();
}
// 创建命名空间
@Test
public void createNameSpace() throws IOException {
// 创建命名空间描述符
NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create("weibo").addConfiguration("ceator", "wang").build();
// 创建命名空间
admin.createNamespace(namespaceDescriptor);
}
5）创建微博内容表
// 创建微博content表
@Test
public void createTableContent() throws IOException {
if(!admin.tableExists(TableName.valueOf(WEIBO_CONTENT))){
// 创建表的描述符
HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(WEIBO_CONTENT));
HColumnDescriptor info = new HColumnDescriptor("info");
info.setMinVersions(1);
info.setMaxVersions(1);
info.setBlockCacheEnabled(true); //是否开启缓存
info.setBlocksize(2048*1024);
tableDescriptor.addFamily(info);
admin.createTable(tableDescriptor);
}
}6）创建用户关系表
* 创建用户关系表
* 方法名 createTableRelations
Table Name weibo:relations
RowKey 用户ID
ColumnFamily attends、fans
ColumnLabel 关注用户ID，粉丝用户ID
ColumnValue 用户ID
Version 1个版本
*/
// 创建relations表
@Test
public void createTableRelations() throws IOException {
if (!admin.tableExists(TableName.valueOf(WEIBO_RELATIONS))) {
HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(WEIBO_RELATIONS));
// 添加列族
HColumnDescriptor attends = new HColumnDescriptor("attends");
attends.setMinVersions(1);
attends.setMaxVersions(1);
attends.setBlockCacheEnabled(true);
attends.setBlocksize(2048*1024);
HColumnDescriptor fans = new HColumnDescriptor("fans");
fans.setMinVersions(1);
fans.setMaxVersions(1);
fans.setBlockCacheEnabled(true);
fans.setBlocksize(2048*1024);
tableDescriptor.addFamily(attends);
tableDescriptor.addFamily(fans);
admin.createTable(tableDescriptor);
}
}7）创建微博收件箱表
/**
* 表结构：
方法名 createTableReceiveContentEmails
Table Name weibo:receive_content_email
RowKey 用户ID
ColumnFamily info
ColumnLabel 用户ID
ColumnValue 取微博内容的RowKey
Version 1000
*/
// 创建email表
@Test
public void createTableEmile() throws IOException {
if(!admin.tableExists(TableName.valueOf(WEIBO_CONTENT_EMAILS))){
HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(WEIBO_CONTENT_EMAILS));
// 创建列族
HColumnDescriptor info = new HColumnDescriptor("info");
info.setMinVersions(1000);
info.setMaxVersions(1000);
info.setBlockCacheEnabled(true);
info.setBlocksize(2018*1024);
// 把列族加入表中
tableDescriptor.addFamily(info);
// 创建表
admin.createTable(tableDescriptor);
}
}8）发布微博内容
a、微博内容表中添加1条数据
b、微博收件箱表对所有粉丝用户添加数据
// 用户发布微博
public void publishWeibo(String uid, String content) throws IOException {
Connection connection = getConnection();
// 1、将发布的内容保存起来 weibo：connect
Table weibo_content = connection.getTable(TableName.valueOf(WEIBO_CONTENT));
// 设定rowkey
long time = System.currentTimeMillis();
String rowkey = uid + "_" + time;
Put put = new Put(rowkey.getBytes());
put.addColumn("info".getBytes(), "content".getBytes(), time, content.getBytes());
weibo_content.put(put);
// 2、获取当前的所有fans查询
Table weibo_relations = connection.getTable(TableName.valueOf(WEIBO_RELATIONS));
Get get = new Get(uid.getBytes());
// 只要fans的列族数据
get.addFamily("fans".getBytes());
Result result = weibo_relations.get(get);
// 如果没有fans 直接结束
if (result.isEmpty()) {
weibo_content.close();
weibo_relations.close();
connection.close();
return;
}
ArrayList<String> uids = new ArrayList<>();
//要是是fans的uid
List<Cell> cells = result.listCells();
for (Cell cell : cells) {
String fans_uid = Bytes.toString(CellUtil.cloneQualifier(cell));
uids.add(fans_uid);
}
// 3、将用户发布的消息rowkey
Table weibo_receive_content_emails = connection.getTable(TableName.valueOf(WEIBO_CONTENT_EMAILS));
for (String fanid : uids) {
Put put1 = new Put(fanid.getBytes());
put1.addColumn("info".getBytes(), uid.getBytes(), time, rowkey.getBytes());
weibo_receive_content_emails.put(put1);
}
weibo_content.close();
weibo_relations.close();
weibo_receive_content_emails.close();
connection.close();
}
9）添加关注用户
// 添加关注用户
public void addAttends(String uid, String... attends) throws IOException {
Connection connection = getConnection();
Table weibo_relations = connection.getTable(TableName.valueOf(WEIBO_RELATIONS));
//1、1的用户保存关注的用户
Put put = new Put(uid.getBytes());
for (String attend : attends) {
put.addColumn("attends".getBytes(), attend.getBytes(), attend.getBytes());
}
weibo_relations.put(put);
//2、被关注的用户同时多了一个fans
for (String attend : attends) {
Put put1 = new Put(attend.getBytes());
put1.addColumn("fans".getBytes(), uid.getBytes(), uid.getBytes());
weibo_relations.put(put1);
}
//3、维护当前用户和被关乎用户的微博查看关系 1关注了23m 此时1就可以看到23m发布的消息
Table weibo_content = connection.getTable(TableName.valueOf(WEIBO_CONTENT));
Table weibo_content_emails = connection.getTable(TableName.valueOf(WEIBO_CONTENT_EMAILS));
ArrayList<String> strings = new ArrayList<>();//用于接收微博rowkey
for (String attend : attends) {
Scan scan = new Scan();
//用于检索当前用户发布微博的rowkey
PrefixFilter prefixFilter = new PrefixFilter((attend + "_").getBytes());
scan.setFilter(prefixFilter);
ResultScanner scanner = weibo_content.getScanner(scan);
//遍历获取微博的rowkey
for (Result result : scanner) {
List<Cell> cells = result.listCells();
for (Cell cell : cells) {
//获取微博rowkey
String weiborowkey = Bytes.toString(CellUtil.cloneRow(cell));
strings.add(weiborowkey);
}
}
}
//把微博内容的rowkey保存到收件箱中
//传入的是当前用户的uid
Put put2 = new Put(uid.getBytes());
for (String weiborowkey : strings) {
//2_13235345436 2_13452346436
String[] split = weiborowkey.split("_");
put2.addColumn("info".getBytes(), split[0].getBytes(), System.currentTimeMillis(), weiborowkey.getBytes());
weibo_content_emails.put(put2);
}
weibo_content.close();
weibo_relations.close();
weibo_content_emails.close();
connection.close();
}10）取消关注用户
//7.取消关注用户
public void cancelAttends(String uid,String... cancelattends) throws IOException {
Connection connection = getConnection();
Table weibo_relations = connection.getTable(TableName.valueOf(WEIBO_RELATIONS));
Delete delete = new Delete(uid.getBytes());
//1.移除当前用户被关注的用户 1. 取消3 M
for (String cancelattend : cancelattends) {
delete.addColumns("attends".getBytes(),cancelattend.getBytes());
}
weibo_relations.delete(delete);
//2.被移除的用户同时取消粉丝
for (String cancelattend : cancelattends) {
//被取消关注的用户，取消粉丝
Delete delete1 = new Delete(cancelattend.getBytes());
delete1.addColumn("fans".getBytes(),uid.getBytes());
weibo_relations.delete(delete1);
}
Table weibo_content_emails = connection.getTable(TableName.valueOf(WEIBO_CONTENT_EMAILS));
//3.删除收件箱中相关的微博rowkey
for (String cancelattend : cancelattends) {
Delete delete2 = new Delete(uid.getBytes());
//删除全部版本
delete2.addColumns("info".getBytes(),cancelattend.getBytes());
weibo_content_emails.delete(delete2);
}
weibo_content_emails.close();
weibo_relations.close();
connection.close();
}11）拉取关注人的微博消息
//8.拉取关注人的微博消息
public void getAttendsContent(String uid) throws IOException {
Connection connection = getConnection();
Table weibo_content_emails = connection.getTable(TableName.valueOf(WEIBO_CONTENT_EMAILS));
//从email表检索关注的用户发布的微博rowkey
Get get = new Get(uid.getBytes());
get.setMaxVersions(5);
Result result = weibo_content_emails.get(get);
//创建一个get集合用于接收get对象
List<Get> gets = new ArrayList<Get>();
List<Cell> cells = result.listCells();
for (Cell cell : cells) {
//获取微博的rowkey
byte[] weiborowkey = CellUtil.cloneValue(cell);
//这个get就是用于在content进行检索数据deget对象
Get get1 = new Get(weiborowkey);
gets.add(get1);
}
//上微博表检索发布的内容
Table weibo_content = connection.getTable(TableName.valueOf(WEIBO_CONTENT));
//获取检索数据的每一个结果集
Result[] results = weibo_content.get(gets);
for (Result result1 : results) {
List<Cell> cells1 = result1.listCells();
for (Cell cell : cells1) {
//打印发布的内容
System.out.println(Bytes.toString(CellUtil.cloneValue(cell)));
}
}
weibo_content.close();
weibo_content_emails.close();
connection.close();
}