需求

陌陌综合案例，业务数据流程图：陌陌用户聊天数据存储到日志log文件中，实时采集到Kafka消息队列，实时消费数据存储到HBase表，最后关联Hive表和Phoenix表进行离线分析和即席查询。

image.png

数据采集：Flume
实时存储：Kafka
离线存储：HBase
离线分析：Hive 离线分析
即时查询：Phoenix 即席查询

• flume

  source源 channel缓存 sinlk 下沉地 目标

• kafka

  producer broker(节点) consumer Consumer Group

  架构 kafka集群 和 zk集群

  broker节点

  Topic partition (leader follower 副本 )

  parititon offset唯一

  sgement 片段

• Hbase 分布式基于hdfs的基于内存的列式存储nosql大表数据库

  主从架构 maser regin server

  zk集群 存放元数据表 所在regin server的地址

  存储设计

  • 逻辑 namespace table rowkey ColumFamly 列族 colum列 value

  • 物理 每个表都有很多分区 rigion 每个分区中每个列族的数据存在Store中

    store分为 memstore(内存) 和 storefile(磁盘)

为什么不直接将Flume的数据给HBase，而统一的给了Kafka，再由Kafka到HBase？

• 避免高并发写导致机器负载过高、实现架构解耦、实现异步高效

• 保证数据一致性

1-Flume实时采集日志到kafka

[图片上传失败...(image-147d44-1660231795046)]

• 编写flume agent 配置文件

• 确定flume的source channel sink

  source : taildir

  channel:内存mm

  sink :kafka sink org.apache.flume.sink.kafka.KafkaSink

一些在kafka生产者设置的代码可以之间在kafka agent中进行设置

列如 kakfk集群文章 acks=1 batch批次大小 topic名 响应时间

需要现在kafka中创建对应的topic 设置分区和副本 先启动kafka然后再运行kafak

# define a1
a1.sources = s1 
a1.channels = c1
a1.sinks = k1

#define s1
a1.sources.s1.type = TAILDIR
#指定一个元数据记录文件
a1.sources.s1.positionFile = /export/data/momo_conf/taildir_momo_kafka.json
#将所有需要监控的数据源变成一个组
a1.sources.s1.filegroups = f1
#指定了f1是谁：监控目录下所有文件
a1.sources.s1.filegroups.f1 = /export/data/momo_data/.*
#指定f1采集到的数据的header中包含一个KV对
a1.sources.s1.headers.f1.type = momo
a1.sources.s1.fileHeader = true

#define c1
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000

#define k1
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.kafka.topic = momo-msg
a1.sinks.k1.kafka.bootstrap.servers = node1.itcast.cn:9092,node2.itcast.cn:9092,node3.itcast.cn:9092
a1.sinks.k1.kafka.flumeBatchSize = 10
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 100

#bind
a1.sources.s1.channels = c1
a1.sinks.k1.channel = c1

• 确定好之后,要在kafka中创建一个topic 指名分区的个数 和副本数

/export/server/kafka/bin/kafka-topics.sh --create \
    --topic momo-msg  \
    --partitions 3 \
    --replication-factor 2 \
    --bootstrap-server node1.itcast.cn:9092,node2.itcast.cn:9092,node3.itcast.cn:9092

• 启动zk 启动kafka

• 启动flume

 /export/server/flume/bin/flume-ng agent \
    -c /export/server/flume/conf/ \
    -n a1 \
    -f /export/data/momo_conf/momo_mem_kafka.properties \
    -Dflume.root.logger=INFO,console

• 总结

kafka sink会自动从header中获取key的值，并以此值最为key发送到broker中。

kafka的分区规则是：

如果key为空，则随机发送到各个分区中。 key不为空，则根据Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions，类似于hash(keyBytes)对分区数取摸，来发送到对应的分区。 所以相同的key肯定会在同一个分区中。

2-JAVA API消费Kafka数据

问题为什么会一批一批的出现数据

这是因为

生产者将数据封装到ProducerRecord中,将ProducerRecord放到一个批次batch中,当该batch达到一定大小,或者一定时间,提交写入请求

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;

/**
 * @description: 将momo的日志从kafka中消费数据,处理完成后写入Hbase
 * @author:
 */
public class MomoKafkaToHbaseText {
 public static void main(String[] args) {
 //todo 1消费kakfa中数据
 /**
 * 采用手动提交分区offset的方式  消费kakfa中的数据
 */
 //1.创建kafka消费者 配置文件
 Properties props = new Properties();
 //需要kafak集群的位置,反序列化类,消费者组,首次消费,自动提交offset 关闭
 props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1.itcast.cn:9092,node2.itcast.cn:9092,node3.itcast.cn:9092");
 //设置key value的序列化类 todo 注意作为消费者 从kafka把字节数组变成String字符串  叫做反序列化
 props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
 props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
 //设置消费者组的编号 todo kafka要求消费者必须属于消费者组
 props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"momo_1");
 //设置消费的位置 从哪里开始消费  合法参数：latest |  earliest(最新的 默认的)
 props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
 //关闭自动提交
 props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

 //2.创建kafkaconsumer对象
 KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
 //3.订阅topic主题
 consumer.subscribe(Collections.singletonList("momo-msg"));
 //4.使用死循环 模拟不间断的拉取数据
 while (true){
 //4-1拉取数据 并设置响应时间
 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMinutes(100));
 //4-2获取每个分区内数据
 Set<TopicPartition> partitions = records.partitions();
 //4-3遍历分区信息
 //记录offset
 long consumerOffset=0L;
 for (TopicPartition partition : partitions) {

 //这时候获取的就是每个分区的数据
 List<ConsumerRecord<String, String>> partRecords = records.records(partition);
 //遍历该分区下所有数据
 for (ConsumerRecord<String, String> record : partRecords) {
 String value = record.value();
 int partIndex = record.partition();
 System.out.println("分区为"+partIndex +"____"+ value);

 //将offset赋值给 consumeroffset
 consumerOffset=record.offset();
 //todo 将kafka中数据写入到Hbase 表中


 }
 //4-4 手动提交分区内offset 消费记录
 //创建map集合用来存储 分区 和 offset
 Map<TopicPartition, OffsetAndMetadata> offset = new HashMap<>();
 offset.put(partition,new OffsetAndMetadata(consumerOffset));
 consumer.commitSync();


 }

 }

 }
}

3-数据实时存储Hbase

Hbase表设计

• rwokey 设计原则

- 唯一原则:一个rowkey对应一条数据
    - 业务原则: 选择**常用的的查询条件**作为前缀
    - 组合原则:将更多的经常查询的条件作为前缀
    - 散列原则:为了避免热点问题,rowkey生成的越散列越好
    - 长度原则: 再不影响业务的情况下,rowkey越端越好
    
    # 查询需求：
     根据【发件人id、收件人id + 消息日期】查询聊天记录
     发件人账号
     时间
     收件人账号（唯一性）
     RowKey = 发件人id + 消息日期 + 收件人id
     列簇：info
     字段：所有字段，20个字段

    # 设计规则：
     业务、唯一、长度、散列、组合
     [唯一性、业务性、热点性（考虑写数据）]

    # 从HBase表查询数据
     1\. RowKey查询最快的：Get
     2\. 前缀匹配查询：Range

    RowKey = 反转(发件人id) + 消息日期 + 收件人id

• habse建表

  habse中建表需要指定 命名空间和 表名 至少一个列族名

  -- 建表
    create 'htbl_momo_msg', {NAME => "info", COMPRESSION => "GZ"}, { NUMREGIONS => 6, SPLITALGO => 'HexStringSplit'}

JAVA API将数据写入habse中

• 构建Hbase连接

    //集中定义成员变量
     private static TableName tableName = TableName.valueOf("htbl_momo_msg");
     private static byte[] cfName = Bytes.toBytes("info");
     private static Table table;
    
     /**
     *  使用静态代码块 完成HBase的连接
     */
     static {
     try {
     // a. hbase配置对象
     Configuration conf = HBaseConfiguration.create();
     conf.set("hbase.zookeeper.quorum", "node1.itcast.cn,node2.itcast.cn,node3.itcast.cn");
     // b. 通过连接工厂类 创建hbase的连接
     Connection conn = ConnectionFactory.createConnection(conf);
    
     // c. 通过连接获取操作的的表的对象
     table = conn.getTable(tableName);
     } catch (IOException e) {
     e.printStackTrace();
     }
     }

• 封装方法

  根据字段信息构建rowkey

  将rowkey 传入put对象 , 实现Table 表的数据增加

 public static void messageTOHbase(String message ) throws Exception {
 // 1.将数据按照分隔符 分隔
 String[] items = message.split("\001");

 //2.根据字段信息设计rowkey
 // RowKey = 反转(发件人id) + 消息日期 + 收件人id    2 0 11
 String senderAccount  = items[2];
 String msgTime = items[0];
 String receiverAccount = items[11];
 String rowKey = StringUtils.reverse(senderAccount)+"_"+msgTime+"_"+receiverAccount;
 //3.构建put对象
 Put put = new Put(rowKey.getBytes());
 // 4\. 添加列信息
 put.addColumn(cfName, Bytes.toBytes("msg_time"),Bytes.toBytes(items[0]));
 put.addColumn(cfName,Bytes.toBytes("sender_nickyname"),Bytes.toBytes(items[1]));
 put.addColumn(cfName,Bytes.toBytes("sender_account"),Bytes.toBytes(items[2]));
 put.addColumn(cfName,Bytes.toBytes("sender_sex"),Bytes.toBytes(items[3]));
 put.addColumn(cfName,Bytes.toBytes("sender_ip"),Bytes.toBytes(items[4]));
 put.addColumn(cfName,Bytes.toBytes("sender_os"),Bytes.toBytes(items[5]));
 put.addColumn(cfName,Bytes.toBytes("sender_phone_type"),Bytes.toBytes(items[6]));
 put.addColumn(cfName,Bytes.toBytes("sender_network"),Bytes.toBytes(items[7]));
 put.addColumn(cfName,Bytes.toBytes("sender_gps"),Bytes.toBytes(items[8]));
 put.addColumn(cfName,Bytes.toBytes("receiver_nickyname"),Bytes.toBytes(items[9]));
 put.addColumn(cfName,Bytes.toBytes("receiver_ip"),Bytes.toBytes(items[10]));
 put.addColumn(cfName,Bytes.toBytes("receiver_account"),Bytes.toBytes(items[11]));
 put.addColumn(cfName,Bytes.toBytes("receiver_os"),Bytes.toBytes(items[12]));
 put.addColumn(cfName,Bytes.toBytes("receiver_phone_type"),Bytes.toBytes(items[13]));
 put.addColumn(cfName,Bytes.toBytes("receiver_network"),Bytes.toBytes(items[14]));
 put.addColumn(cfName,Bytes.toBytes("receiver_gps"),Bytes.toBytes(items[15]));
 put.addColumn(cfName,Bytes.toBytes("receiver_sex"),Bytes.toBytes(items[16]));
 put.addColumn(cfName,Bytes.toBytes("msg_type"),Bytes.toBytes(items[17]));
 put.addColumn(cfName,Bytes.toBytes("distance"),Bytes.toBytes(items[18]));
 put.addColumn(cfName,Bytes.toBytes("message"),Bytes.toBytes(items[19]));
 // 5.调用Table对象的put方法 将数据插入到HBase中
 table.put(put);

 }
}

4-hive集成habse

注意使用hive集成habse时

• 建表注意事项

  hive一点是外部表

  hive表 以hfile存储

  hive字段映射一 一对应, 列族:列名

  指定rowkey :key

  指定表名 tblproperties('hbase.table.name'='htbl_momo_msg');

CREATE EXTERNAL TABLE IF NOT EXISTS tbl_momo_msg (
 id string,
 msg_time string ,
 sender_nickyname string ,
 sender_account string ,
 sender_sex string ,
 sender_ip string ,
 sender_os string ,
 sender_phone_type string ,
 sender_network string ,
 sender_gps string ,
 receiver_nickyname string ,
 receiver_ip string ,
 receiver_account string ,
 receiver_os string ,
 receiver_phone_type string ,
 receiver_network string ,
 receiver_gps string ,
 receiver_sex string ,
 msg_type string ,
 distance string ,
 message string
) 
stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
with serdeproperties('hbase.columns.mapping'=':key,info:msg_time,info:sender_nickyname,info:sender_account,info:sender_sex,info:sender_ip,info:sender_os,info:sender_phone_type,info:sender_network,info:sender_gps,info:receiver_nickyname,info:receiver_ip,info:receiver_account,info:receiver_os,info:receiver_phone_type,info:receiver_network,info:receiver_gps,info:receiver_sex,info:msg_type,info:distance,info:message') tblproperties('hbase.table.name'='htbl_momo_msg');

5-Phoenix 即席查询

使用Phoenix关联Hbase实现即时查询

[图片上传失败...(image-4d0520-1660231795043)]

• 1、服务启动

  
  
  image.png

• 2、创建视图view，关联HBase表

CREATE VIEW IF NOT EXISTS "htbl_momo_msg" (
   "rk" varchar primary key,
   "info"."msg_time" varchar ,
   "info"."sender_nickyname" varchar ,
   "info"."sender_account" varchar ,
   "info"."sender_sex" varchar ,
   "info"."sender_ip" varchar ,
   "info"."sender_os" varchar ,
   "info"."sender_phone_type" varchar ,
   "info"."sender_network" varchar ,
   "info"."sender_gps" varchar ,
   "info"."receiver_nickyname" varchar ,
   "info"."receiver_ip" varchar ,
   "info"."receiver_account" varchar ,
   "info"."receiver_os" varchar ,
   "info"."receiver_phone_type" varchar ,
   "info"."receiver_network" varchar ,
   "info"."receiver_gps" varchar ,
   "info"."receiver_sex" varchar ,
   "info"."msg_type" varchar ,
   "info"."distance" varchar ,
   "info"."message" varchar
  );

• 3、即时查询

-- 基础查询
    SELECT
     "rk", "info"."sender_account", "info"."receiver_account"
    FROM "htbl_momo_msg"
    LIMIT 10;
    
    -- 查询条目数
    SELECT COUNT(1) AS TOTAL FROM "htbl_momo_msg" ;
    
    -- 查询每个发送人发送的消息数
    SELECT
     "info"."sender_account" ,
     COUNT(1) AS TOTAL
    FROM "htbl_momo_msg"
    GROUP BY "info"."sender_account"
    ORDER BY TOTAL DESC
    LIMIT 10;
    
    -- 查询每个发送人聊天的人数
    SELECT
     "info"."sender_account" ,
     COUNT(DISTINCT "info"."receiver_account") AS TOTAL
    FROM "htbl_momo_msg"
    GROUP BY "info"."sender_account"
    ORDER BY total DESC 
    LIMIT 10;

在实际项目中，为了即席查询更快，可以建议索引表，支持：全局索引、覆盖索引和本地索引。

附录-maven依赖

 <!--远程仓库-->
 <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>

 <!--kafka 客户端-->
 <dependency>
 <groupId>org.apache.kafka</groupId>
 <artifactId>kafka-clients</artifactId>
 <version>2.4.1</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></pre>