❤️让人心跳加速的陌陌案例，大数据必需学会的基础案例！❤️ 【推荐收藏】

Java
467
0
0
2022-10-09
2021年大数据HBase（五）：HBase的相关操作-JavaAPI方式！【建议收藏】
上一篇
Maven最新版的下载与安装教程（详细教程）
下一篇
标签   HBase 大数据

陌陌案例

一、陌陌案例的需求说明

需求: 将陌陌中聊天记录存储到hbase中, 并提供查询的方案

数据特点: 需要高并发写入操作, 读取操作较少 写多读少场景

img

二、陌陌案例中表设计内容

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的表的压缩方案的选择

img

由于数据 是写多 读少的场景, 基本上 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个

建表操作:

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中创建存储数据的表:

create 'MOMO_CHAT:MSG' ,{NAME=>'C1',COMPRESSION=>'GZ'},{NUMREGIONS=>6 , SPLITALGO=>'HexStringSplit'}

2) 创建maven项目,加载pom依赖:

 <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 在 资料的 陌陌海量消息存储案例目录下

img

4) 创建相关的包结构:

  • 存储工具类: com.it.momo_chat.utils 
  • 存储实体类: com.it.momo_chat.entity
  • 存储接口类: com.it.momo_chat.service
  • 存储服务类: com.it.momo_chat.service.impl

5) 导入相关的工具类和实体类 : 在 资料的 陌陌海量消息存储案例目录下

img

2、生成10w条数据

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）接口程序

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）实现类

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 ;
    }
​
}
​
  • 📢博客主页：https://lansonli.blog.csdn.net
  • 📢欢迎点赞 👍 收藏 ⭐留言 📝 如有错误敬请指正！
  • 📢本文由 Lansonli 原创，首发于 CSDN博客🙉
  • 📢大数据系列文章会每天更新，停下休息的时候不要忘了别人还在奔跑，希望大家抓紧时间学习，全力奔赴更美好的生活✨
2021年大数据HBase（五）：HBase的相关操作-JavaAPI方式！【建议收藏】
上一篇
Maven最新版的下载与安装教程（详细教程）
下一篇