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

Lansonli

发布于 2021-10-11 16:33:48

7300

发布于 2021-10-11 16:33:48

文章被收录于专栏：Lansonli技术博客Lansonli技术博客

陌陌案例

一、陌陌案例的需求说明

需求: 将陌陌中聊天记录存储到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个

建表操作:

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

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条数据

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 ;
    }

}