陌陌案例
一、陌陌案例的需求说明
需求: 将陌陌中聊天记录存储到hbase中, 并提供查询的方案
数据特点: 需要高并发写入操作, 读取操作较少 写多读少场景
二、陌陌案例中表设计内容
1、 hbase的名称空间(命名空间)
hbase的名称空间, 可以将其理解为MySQL中数据库
思考: MySQL中为什么要有这个数据库概念, 有什么作用呢?
因为: 通过库将相关类型表放置在一起, 方便管理 可以基于库进行权限管理工作
同样, 对于hbase来讲, 也需要有类似这样功能, 这个时候, hbase推出 名称空间, 可以通过在hbase中构建多个名称空间, 将表放置在不同的名称空间下, 进行分别管理操作
注意:
- hbase默认提供了两个名称空间: default 和 hbase
- default: 默认名称空间, 当我们创建表的时候, 没有指定名称空间, 默认就是创建到这个default空间下
- hbase: 系统的名称空间, 主要是用于存储系统相关的表 meta表(元数据表) ,一般不使用
如何操作hbase的名称空间:
1) 如何创建名称空间 格式: create_namespace '名称空间名称' 2) 如何查看名称空间 格式: 查看所有的名称空间: list_namespace 查看某一个名称空间: describe_namespace '名称空间名称' 3) 如何在指定的名称空间下, 创建表 格式: create '名称空间:表名' ,'列族1'... 4) 如何删除名称空间 格式: drop_namespace '名称空间' 注意 如果对应空间下, 还有表, 是无法删除, 必须先删除表
2、 hbase表的列族的设计
能少则少, 能用一个解决的, 坚决不使用两个
官方建议: 一般列族的配置 不大于 5个 支持非常多
本次陌陌案例采用一个列族来解决: C1
3、hbase表的版本设计
版本设计: 是否需要存储历史变更记录, 或者说数据是否会有历史变更操作
思考: 陌陌案例聊天, 是否会存在变更呢? 不会发生变更, 所以版本设置为 1(默认即可)
4、hbase的表的压缩方案的选择
由于数据 是写多 读少的场景, 基本上 90%以上都是写操作, 而且数据量非常的大, 希望能够在有限的空间下, 存储更多的数据, 此时可以选修压缩比最高的: GZIP(GZ)
如果 读的多, 而且数据量比较大, 可以采用 LZO 或者snappy
如何设置压缩方案
在创建表时指定压缩方案: create '表名' , {NAME='列族',COMPRESSION=>'压缩方案'} 给以及建好的表添加压缩方案: alter '表名' , {NAME='列族',COMPRESSION=>'压缩方案'} 案例: create 'MOMO_CHAT:MSG',{NAME=>'C1',COMPRESSION=>'GZ'}
5、hbase表的预分区
默认情况下, 创建一个表 只有一个region,而一个region只能被一个regionServer所管理, 一个regionServer读写性能有限,而且hbase集群一般由一些廉价的服务器组建集群
如果此时需要对这个表进行大量的读写操作, 最终这些读写请求, 全部负载给某一个台regionServer上, 由于单台节点负载并不是特别高, 有可能会导致读写性能急剧下降, 甚至宕机的风险
请问如何解决呢? 如果这些并发请求, 能够负载到各个regionServer上, 问题就可以解决了,但是一个region依然无法办到
解决方案: 在建表的时候, 指定表的region的数量, 让其能够一次性预先的拥有多个region, 而多个region可以负载到各个regionServer上, 然后在进行读写操作的时候, 就可以将并发的请求落在各个regionServer上
而这种解决方案, 就是HBase的预分区 :
目的: 在建表直接产生多个region
hbase是通过对rowkey的范围, 对region进行划分, 每个region都会有起始的rowkey 和 结束rowkey表示这个region所存储数据范围, 在插入数据时候, 如果rowkey在某一个region的范围, 那么直接将数据插入到这个region中
默认情况下: 一个表只有一个region , name这个region的范围是什么呢? startkey: '' , endkey: '' 如果我以 : 1, 2 ,3 ,4 5 划分一个个region, 请思考有几个region呢? 6 '' ~ 1 1 ~ 2 2 ~ 3 3 ~ 4 4 ~ 5 5 ~ ''
如何设置hbase的预分区呢?
方式一: 手动分区 格式: create '表名' ,'列族1'... , SPLITS=>['1','2','3','4','5'] 方式二: 通过读取一个外部的文件, 来划分region 格式: create '表名','列族1' ...., SPLITS_FILE => '文件路径' 方式三: hash 16进制 分区方案 create '表名' ,'列族名称1', .... , {NUMREGIONS=>N , SPLITALGO=>'HexStringSplit'}
本次陌陌案例, 将会采用 hash 16进制分区方案 : 分区的数量一般为regionServer数量的倍数 设置 6个
建表操作:
代码语言：javascript复制create 'MOMO_CHAT:MSG' ,{NAME=>'C1',COMPRESSION=>'GZ'},{NUMREGIONS=>6 , SPLITALGO=>'HexStringSplit'}思考: 是否只需要设置预分区, 就一定可以保证让所有的数据都均匀落在不同region中呢? 不是的
6、hbase的中rowkey的设计原则
官方rowkey的设置建议要求:
1) 避免使用递增行键/时序数据 当做rowkey的前缀 因为: 递增行键或者时序数据, 前面数字有可能是一成不变, 此时会出现数据热点问题(所有数据都跑到一个region中) 2) 避免rowkey和列的长度过大(长) 因为: 希望数据能够在内存中保留的越多, 读取的效率越高, 如果rowkey或者列设置比较长, 导致在有限内存中存储数据更小, 从而让数据提前的就flush磁盘上, 影响读取效率 建议: rowkey长度一般为 10~100字节左右 , 尽可能的越短越好 3) 使用Long类型比String类型更节省空间: 如果rowkey中都是数字, 建议使用Long获取其他数值类型 4) 保证rowkey的唯一性
如何避免热点问题:
1) 反转策略: 比如说可以将手机号 或者 时间戳等 这种前面一样但是后面会呈现随机的数据, 进行反转工作 就可以保证rowkey的前缀都不尽相同, 从而让数据能够落在不同的region中 2) 加盐策略: 给rowkey前缀添加固定长度的随机数 , 来保证让数据落在不同region中 3) hash取模: 给相同的数据加上同样的盐, 从而保证相关联的数据都在一起, 也可以保证数据落在笔筒region中
在陌陌案例中, 如何设计rowkey呢? 以查询作为参考点, 决定你的rowkey应该放什么数据
HASH(MD5加密)_发件人账户_收件人账户_时间戳 通过 HASH(MD5加密) 可以确保数据均匀落在不同region上, 同时也可以保证 同一对发件人和收件人都存储在一个region中
三.、陌陌案例实现
1、准备工作
1) 在hbase中创建存储数据的表:
代码语言：javascript复制create 'MOMO_CHAT:MSG' ,{NAME=>'C1',COMPRESSION=>'GZ'},{NUMREGIONS=>6 , SPLITALGO=>'HexStringSplit'}2) 创建maven项目,加载pom依赖:
代码语言：javascript复制 <repositories>
<repository>
<id>aliyun</id>
<url>http://maven.aliyun.com/nexus/content/groups/public/</url>
<releases>
<enabled>true</enabled>
</releases>
<snapshots>
<enabled>false</enabled>
<updatePolicy>never</updatePolicy>
</snapshots>
</repository>
</repositories>
<dependencies>
<!--Hbase 客户端-->
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>2.1.0</version>
</dependency>
<!--poi包: 用于java读取Excel文件中数据包-->
<dependency>
<groupId>com.github.cloudecho</groupId>
<artifactId>xmlbean</artifactId>
<version>1.5.5</version>
</dependency>
<dependency>
<groupId>org.apache.poi</groupId>
<artifactId>poi</artifactId>
<version>4.0.1</version>
</dependency>
<dependency>
<groupId>org.apache.poi</groupId>
<artifactId>poi-ooxml</artifactId>
<version>4.0.1</version>
</dependency>
<dependency>
<groupId>org.apache.poi</groupId>
<artifactId>poi-ooxml-schemas</artifactId>
<version>4.0.1</version>
</dependency>
<!--json数据包 : json本质上就是有一定格式字符串-->
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.62</version>
</dependency>
<!--Phoenix相关jar包 可以省略(如果报错)-->
<!--<dependency>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix-core</artifactId>
<version>5.0.0-HBase-2.0</version>
</dependency>
<dependency>
<groupId>org.apache.phoenix</groupId>
<artifactId>phoenix-queryserver-client</artifactId>
<version>5.0.0-HBase-2.0</version>
</dependency>-->
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<target>1.8</target>
<source>1.8</source>
</configuration>
</plugin>
</plugins>
</build>3) 导入相关的配置文件 : log4j.properties 在 资料的 陌陌海量消息存储案例目录下
4) 创建相关的包结构:
- 存储工具类: com.it.momo_chat.utils
- 存储实体类: com.it.momo_chat.entity
- 存储接口类: com.it.momo_chat.service
- 存储服务类: com.it.momo_chat.service.impl
5) 导入相关的工具类和实体类 : 在 资料的 陌陌海量消息存储案例目录下
2、生成10w条数据
代码语言：javascript复制package com.it.momo_chat.utils;
import com.it.momo_chat.entity.Msg;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.MD5Hash;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Map;
public class Gen {
public static void main(String[] args) throws Exception {
//1. 读取数据:
String xlxsPath = "D:\传智工作\上课\北京大数据48期\实时阶段课程\day16_实时阶段_HBase\资料\陌陌海量消息存储案例\测试数据集.xlsx";
Map<String, List<String>> resultMap = ExcelReader.readXlsx(xlxsPath, "陌陌数据");
//4. 写入到Hbase中:
//4.1: 根据Hbase的连接工厂, 创建Hbase的连接对象
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");
Connection connection = ConnectionFactory.createConnection(conf);
//4.2: 根据连接对象, 获取管理对象: Table
Table table = connection.getTable(TableName.valueOf("MOMO_CHAT:MSG"));
//3. 生成10w条数据
for(int i =0 ; i<100000 ; i ){
//2. 调用 randomRow方法, 随机生成一行数据
Msg rowData = randomRow(resultMap);
//4.3: 执行相关的操作: 添加数据
Put put = new Put(getRowkey(rowData));
put.addColumn("C1".getBytes(),"msg_time".getBytes(),rowData.getMsg_time().getBytes());
put.addColumn("C1".getBytes(),"sender_nickyname".getBytes(),rowData.getSender_nickyname().getBytes());
put.addColumn("C1".getBytes(),"sender_account".getBytes(),rowData.getSender_account().getBytes());
put.addColumn("C1".getBytes(),"sender_sex".getBytes(),rowData.getSender_sex().getBytes());
put.addColumn("C1".getBytes(),"sender_ip".getBytes(),rowData.getSender_ip().getBytes());
put.addColumn("C1".getBytes(),"sender_os".getBytes(),rowData.getSender_os().getBytes());
put.addColumn("C1".getBytes(),"sender_phone_type".getBytes(),rowData.getSender_phone_type().getBytes());
put.addColumn("C1".getBytes(),"sender_network".getBytes(),rowData.getSender_network().getBytes());
put.addColumn("C1".getBytes(),"sender_gps".getBytes(),rowData.getSender_gps().getBytes());
put.addColumn("C1".getBytes(),"receiver_nickyname".getBytes(),rowData.getReceiver_nickyname().getBytes());
put.addColumn("C1".getBytes(),"receiver_ip".getBytes(),rowData.getReceiver_ip().getBytes());
put.addColumn("C1".getBytes(),"receiver_account".getBytes(),rowData.getReceiver_account().getBytes());
put.addColumn("C1".getBytes(),"receiver_os".getBytes(),rowData.getReceiver_os().getBytes());
put.addColumn("C1".getBytes(),"receiver_phone_type".getBytes(),rowData.getReceiver_phone_type().getBytes());
put.addColumn("C1".getBytes(),"receiver_network".getBytes(),rowData.getReceiver_network().getBytes());
put.addColumn("C1".getBytes(),"receiver_gps".getBytes(),rowData.getReceiver_gps().getBytes());
put.addColumn("C1".getBytes(),"receiver_sex".getBytes(),rowData.getReceiver_sex().getBytes());
put.addColumn("C1".getBytes(),"msg_type".getBytes(),rowData.getMsg_type().getBytes());
put.addColumn("C1".getBytes(),"distance".getBytes(),rowData.getDistance().getBytes());
put.addColumn("C1".getBytes(),"message".getBytes(),rowData.getMessage().getBytes());
table.put(put);
System.out.println("数据生成到-->" i);
}
//4.4: 释放资源
table.close();
connection.close();
}
//定义一个方法: 随机生成一行数据
public static Msg randomRow(Map<String, List<String>> resultMap){
Msg msg = new Msg();
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date date = new Date();
msg.setMsg_time(format.format(date));
msg.setSender_nickyname(ExcelReader.randomColumn(resultMap,"sender_nickyname"));
msg.setSender_account(ExcelReader.randomColumn(resultMap,"sender_account"));
msg.setSender_sex(ExcelReader.randomColumn(resultMap,"sender_sex"));
msg.setSender_ip(ExcelReader.randomColumn(resultMap,"sender_ip"));
msg.setSender_os(ExcelReader.randomColumn(resultMap,"sender_os"));
msg.setSender_phone_type(ExcelReader.randomColumn(resultMap,"sender_phone_type"));
msg.setSender_network(ExcelReader.randomColumn(resultMap,"sender_network"));
msg.setSender_gps(ExcelReader.randomColumn(resultMap,"sender_gps"));
msg.setReceiver_nickyname(ExcelReader.randomColumn(resultMap,"receiver_nickyname"));
msg.setReceiver_ip(ExcelReader.randomColumn(resultMap,"receiver_ip"));
msg.setReceiver_account(ExcelReader.randomColumn(resultMap,"receiver_account"));
msg.setReceiver_os(ExcelReader.randomColumn(resultMap,"receiver_os"));
msg.setReceiver_phone_type(ExcelReader.randomColumn(resultMap,"receiver_phone_type"));
msg.setReceiver_network(ExcelReader.randomColumn(resultMap,"receiver_network"));
msg.setReceiver_gps(ExcelReader.randomColumn(resultMap,"receiver_gps"));
msg.setReceiver_sex(ExcelReader.randomColumn(resultMap,"receiver_sex"));
msg.setMsg_type(ExcelReader.randomColumn(resultMap,"msg_type"));
msg.setDistance(ExcelReader.randomColumn(resultMap,"distance"));
msg.setMessage(ExcelReader.randomColumn(resultMap,"message"));
return msg;
}
// 生成 rowkey
private static byte[] getRowkey(Msg msg) throws ParseException {
// 3. 构建ROWKEY
// 发件人ID1反转
StringBuilder stringBuilder = new StringBuilder(msg.getSender_account());
stringBuilder.append("_");
stringBuilder.append(msg.getReceiver_account());
stringBuilder.append("_");
// 转换为时间戳
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
stringBuilder.append(sdf.parse(msg.getMsg_time()).getTime());
byte[] orginkey = Bytes.toBytes(stringBuilder.toString());
// 为了避免ROWKEY过长，取前八位
String md5AsHex = MD5Hash.getMD5AsHex(orginkey).substring(0, 8);
return Bytes.toBytes(md5AsHex "_" stringBuilder.toString());
}
}3、查询操作
需求: 请提供一个功能, 能够根据指定的日期(某一天)和收件人账户以及发件人账户 查询 消息数据
1）接口程序
代码语言：javascript复制package com.it.momo_chat.service;
import com.it.momo_chat.entity.Msg;
import java.util.List;
public interface ChatMessageService {
public List<Msg> getMessage(String date , String sender ,String receiver) throws Exception;
public void close() throws Exception;
}2）实现类
代码语言：javascript复制package com.it.momo_chat.service.impl;
import com.it.momo_chat.entity.Msg;
import com.it.momo_chat.service.ChatMessageService;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import java.util.ArrayList;
import java.util.List;
public class ChatMessageServiceImpl implements ChatMessageService {
private Connection connection ;
private Table table;
@Override
public List<Msg> getMessage(String date, String sender, String receiver) throws Exception {
//1. 根据连接工厂, 创建连接对象
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum","node1:2181,node2:2181,node3:2181");
connection = ConnectionFactory.createConnection(conf);
//2. 根据连接对象, 获取管理对象: Table对象
table = connection.getTable(TableName.valueOf("MOMO_CHAT:MSG"));
//3. 执行相关的操作
Scan scan = new Scan();
scan.setLimit(100);
String startDate = date " 00:00:00";
String endDate = date " 23:59:59";
SingleColumnValueFilter startMsg_filter = new SingleColumnValueFilter("C1".getBytes(), "msg_time".getBytes(),
CompareOperator.GREATER_OR_EQUAL, new BinaryComparator(startDate.getBytes()));
SingleColumnValueFilter endMsg_filter = new SingleColumnValueFilter("C1".getBytes(), "msg_time".getBytes(),
CompareOperator.LESS_OR_EQUAL, new BinaryComparator(endDate.getBytes()));
SingleColumnValueFilter senderMsg_filter = new SingleColumnValueFilter("C1".getBytes(), "sender_account".getBytes(),
CompareOperator.EQUAL, new BinaryComparator(sender.getBytes()));
SingleColumnValueFilter receiverMsg_filter = new SingleColumnValueFilter("C1".getBytes(), "receiver_account".getBytes(),
CompareOperator.EQUAL, new BinaryComparator(receiver.getBytes()));
FilterList filterList = new FilterList();
filterList.addFilter(startMsg_filter);
filterList.addFilter(endMsg_filter);
filterList.addFilter(senderMsg_filter);
filterList.addFilter(receiverMsg_filter);
scan.setFilter(filterList);
ResultScanner results = table.getScanner(scan);
//4. 处理结果集
List<Msg> msgList = new ArrayList<Msg>();
for (Result result : results) {
List<Cell> listCells = result.listCells();
// 封装每一行数据, 返回一个msg对象
Msg msg = resusltMsg( listCells);
msgList.add(msg);
}
//5. 釋放資源
close();
return msgList;
}
@Override
public void close() throws Exception {
table.close();
connection.close();
}
private Msg resusltMsg( List<Cell> listCells) {
Msg msg = new Msg();
for (Cell cell : listCells) {
byte[] qualifierBytes = CellUtil.cloneQualifier(cell);
String qualifier = Bytes.toString(qualifierBytes);
byte[] valueBytes = CellUtil.cloneValue(cell);
String value = Bytes.toString(valueBytes);
if("msg_time".equalsIgnoreCase(qualifier)){
msg.setMsg_time(value);
}
if("sender_nickyname".equalsIgnoreCase(qualifier)){
msg.setSender_nickyname(value);
}
if("sender_account".equalsIgnoreCase(qualifier)){
msg.setSender_account(value);
}
if("sender_sex".equalsIgnoreCase(qualifier)){
msg.setSender_sex(value);
}
if("sender_ip".equalsIgnoreCase(qualifier)){
msg.setSender_ip(value);
}
if("sender_os".equalsIgnoreCase(qualifier)){
msg.setSender_os(value);
}
if("sender_phone_type".equalsIgnoreCase(qualifier)){
msg.setSender_phone_type(value);
}
if("sender_network".equalsIgnoreCase(qualifier)){
msg.setSender_network(value);
}
if("sender_gps".equalsIgnoreCase(qualifier)){
msg.setSender_gps(value);
}
if("receiver_nickyname".equalsIgnoreCase(qualifier)){
msg.setReceiver_nickyname(value);
}
if("receiver_ip".equalsIgnoreCase(qualifier)){
msg.setReceiver_ip(value);
}
if("receiver_account".equalsIgnoreCase(qualifier)){
msg.setReceiver_account(value);
}
if("receiver_os".equalsIgnoreCase(qualifier)){
msg.setReceiver_os(value);
}
if("receiver_phone_type".equalsIgnoreCase(qualifier)){
msg.setReceiver_phone_type(value);
}
if("receiver_network".equalsIgnoreCase(qualifier)){
msg.setReceiver_network(value);
}
if("receiver_gps".equalsIgnoreCase(qualifier)){
msg.setReceiver_gps(value);
}
if("receiver_sex".equalsIgnoreCase(qualifier)){
msg.setReceiver_sex(value);
}
if("msg_type".equalsIgnoreCase(qualifier)){
msg.setMsg_type(value);
}
if("distance".equalsIgnoreCase(qualifier)){
msg.setDistance(value);
}
if("message".equalsIgnoreCase(qualifier)){
msg.setMessage(value);
}
}
return msg ;
}
}
