导语 | 本文描述了如何在腾讯云上使用云化后大数据组件来完成实时分析系统的设计和实现，阅读过程中通过对比云Ckafka、Flink和MySQL等组件的使用差异来体现云化方案的优势。文中以视频直播礼物打赏的场景为例，展示全/半托管服务下开发的便利，便于读者对视频直播系统的设计有一个初步了解。
一、解决方案描述
（一）概述
本方案结合腾讯云CKafka、流计算Oceanus、私有网络VPC、商业智能分析BI等，对视频直播行业数字化运营进行实时可视化分析。分析指标包含观看直播人员的地区分布、各级别会员统计、各模块打赏礼物情况、在线人数等。
视频直播场景
（二）方案架构及优势
根据以上视频直播场景，设计了如下架构图：
架构图
涉及产品列表：
- 流计算Oceanus
- 私有网络VPC
- 消息队列CKafka
- 云数据库MySQL
- EMR集群HBase组件
- 商业智能分析服务
二、前置准备
购买并创建相应的大数据组件。
（一）创建VPC私有网络
私有网络是一块您在腾讯云上自定义的逻辑隔离网络空间，在构建MySQL、EMR，ClickHouse集群等服务时选择的网络必须保持一致，网络才能互通。否则需要使用对等连接、VPN等方式打通网络。
（页面地址：https://console.cloud.tencent.com/vpc/vpc?rid=8）
（二）创建Oceanus集群
流计算Oceanus服务兼容原生的Flink任务。在 Oceanus 控制台的【集群管理->【新建集群】页面创建集群，选择地域、可用区、VPC、日志、存储，设置初始密码等。VPC及子网使用刚刚创建好的网络。创建完后Flink的集群如下：
Oceanus集群
（三）创建消息队列Ckafka
消息队列CKafka（Cloud Kafka）是基于开源Apache Kafka消息队列引擎，提供高吞吐性能、高可扩展性的消息队列服务。消息队列CKafka完美兼容Apache kafka0.9、0.10、1.1、2.4、2.8版本接口，在性能、扩展性、业务安全保障、运维等方面具有超强优势，让您在享受低成本、超强功能的同时，免除繁琐运维工作。
（页面地址：https://cloud.tencent.com/product/ckafka）
- 创建Ckafka集群
注意私有网络和子网选择之前创建的网络和子网：
Kafka集群
- 创建topic
创建topic
- 模拟发送数据到 topic
- kafka客户端
进入同子网的CVM下，启动kafka客户端，模拟发送数据，具体操作参考文档：
（https://cloud.tencent.com/document/product/597/56840）
- 使用脚本发送
脚本一：Java参考以下官方网址：
（https://cloud.tencent.com/document/product/597/54834）
脚本二：Python脚本生成模拟数据：
#!/usr/bin/python3# 首次使用该脚本，需 "pip3 install kafka" 安装kafka模块import jsonimport randomimport timefrom kafka import KafkaProducer
TIME_FORMAT = "%Y-%m-%d %H:%M:%S"PROVINCES = ["北京", "广东", "山东", "江苏", "河南", "上海", "河北", "浙江", "香港", "陕西", "湖南", "重庆", "福建", "天津", "云南", "四川", "广西", "安徽", "海南", "江西", "湖北", "山西", "辽宁", "台湾", "黑龙江", "内蒙古", "澳门", "贵州", "甘肃", "青海", "新疆", "西藏", "吉林", "宁夏"]
broker_lists = ['172.28.28.13:9092']topic_live_gift_total = 'live_gift_total'topic_live_streaming_log = 'live_streaming_log'
producer = KafkaProducer(bootstrap_servers=broker_lists, value_serializer=lambda m: json.dumps(m).encode('ascii'))
# 模拟几天前，几小时前的数据pre_day_count = 0pre_hour_count = 0hour_unit = 3600day_unit = 3600 * 24
def generate_data_live_gift_total(): # construct time update_time = time.time() - day_unit * pre_day_count update_time_str = time.strftime(TIME_FORMAT, time.localtime(update_time)) create_time = update_time - hour_unit * pre_hour_count create_time_str = time.strftime(TIME_FORMAT, time.localtime(create_time)) results = []
for _ in range(0, 10): user_id = random.randint(2000, 4000) random_gift_type = random.randint(1, 10) random_gift_total = random.randint(1, 100) msg_kv = {"user_id": user_id, "gift_type": random_gift_type, "gift_total_amount": random_gift_total, "create_time": create_time_str, "update_time": update_time_str} results.append(msg_kv) return results
def generate_live_streaming_log(): # construct time update_time = time.time() - day_unit * pre_day_count leave_time_str = time.strftime(TIME_FORMAT, time.localtime(update_time)) create_time = update_time - hour_unit * pre_hour_count create_time_str = time.strftime(TIME_FORMAT, time.localtime(create_time)) results = []
for _ in range(0, 10): user_id = random.randint(2000, 4000) random_province = random.randint(0, len(PROVINCES) - 1) province_name = PROVINCES[random_province] grade = random.randint(1, 5) msg_kv = {"user_id": user_id, "ip": "123.0.0." str(user_id % 255), "room_id": 20210813, "arrive_time": create_time_str, "create_time": create_time_str, "leave_time": leave_time_str, "region": 1122, "grade": (user_id % 5 1), "province": province_name} results.append(msg_kv) return results
def send_data(topic, msgs): count = 0
# produce asynchronously for msg in msgs: import time time.sleep(1) count = 1 producer.send(topic, msg) print(" send %d data...n %s" % (count, msg))
producer.flush()
if __name__ == '__main__': count = 1 while True: time.sleep(60) #for _ in range(count): msg_live_stream_logs = generate_live_streaming_log() send_data(topic_live_streaming_log, msg_live_stream_logs)
msg_topic_live_gift_totals = generate_data_live_gift_total() send_data(topic_live_gift_total, msg_topic_live_gift_totals)（四）创建EMR集群
EMR是云端托管的弹性开源泛Hadoop服务，支持Spark、HBase、Presto、Flink、Druid等大数据框架，本次示例主要需要使用Flume、Hive、YARN、HUE、Oozie组件。
（页面地址https://console.cloud.tencent.com/emr）
- 在EMR集群中安装HBase组件。
HBase组件
- 如果生产环境，服务器配置可根据实际情况选择，示例中选择了低配服务器，网络需要选择之前创建好的VPC网络，始终保持服务组件在同一VPC下。
网络选择
- 进入HBase Master节点
HBaseMaster节点
- 点击登录进入服务器
- 创建Hbase表
# 进入HBase命令[root@172~]# hbase shell# 建表语句create ‘dim_hbase’, ‘cf’（五）创建云数据库MySQL
云数据库MySQL（TencentDB for MySQL）是腾讯云基于开源数据库MySQL专业打造的高性能分布式数据存储服务，让用户能够在云中更轻松地设置、操作和扩展关系数据库。
（页面地址：https://console.cloud.tencent.com/cdb）
新建MySQL服务的页面需要注意选择的网络是之前创建好的：
MySQL创建
创建完MySQL服务后，需要修改binlog参数，如图修改为FULL（默认值为MINIMAL）
修改参数
修改完参数后，登陆MySQL创建示例所需要的数据库和数据库表。
- 登陆MySQL云数据库
登录
- 新建数据库
打开SQL窗口或可视化页面创建数据库和表：
CREATE DATABASE livedb; --创建数据库列表（六）创建商业智能分析
商业智能分析（Business Intelligence，BI）支持自服务数据准备、探索式分析和企业级管控，是新一代的敏捷自助型BI服务平台。只需几分钟，您就可以在云端轻松自如地完成数据分析、业务数据探查、报表制作等一系列数据可视化操作。便捷的拖拉拽式交互操作方式，让您无需依赖IT人员，无需担心试错成本，快速洞察数据背后的关联、趋势和逻辑。
（页面地址：https://cloud.tencent.com/product/bi）
- 购买商业智能分析
- 需要主账号购买资源，购买时需根据创建的子账号数来进行购买。
BI购买
- 子用户提出申请：
子用户申请
- 主账号审核通过。并给子用户授予添加数据源，创建数据集，查看报告的权限。
- 添加MySQL数据源
这里选用开启外网方式连接，更多连接方式见：
（https://cloud.tencent.com/document/product/590/19294）
- 打开购买的MySQL实例，开启外网：
MySQL开启外网
- 将SaaS BI（119.29.66.144:3306）添加到MySQL数据库安全组
添加安全组1
添加安全组2
注意添加的是MySQL3306端口，不是外网映射的端口。
添加安全组3
- 创建MySQL账户并配置权限
创建账户，并设置账号密码，注意主机IP设置为%：
创建账户1
创建账户2
设置账号权限：
设置权限1
设置权限2
- 进入智能商业分析，连接MySQL数据库。添加数据源->MySQL，填写完成后点击测试连接。
三、方案实现
接下来通过案例为您介绍如何利用流计算服务Oceanus实现视频直播数字化运营的实时可视化数据处理与分析。
（一）解决方案
- 业务目标
这里只列取以下3种统计指标：
- 全站观看直播用户分布；
- 礼物总和统计各模块；
- 礼物统计源数据格式。
事件log：live_streaming_log（topic）：
Ckafka内部采用json格式存储，展现出来的数据如下所示：
{'user_id': 3165, 'ip': '123.0.0.105', 'room_id': 20210813, 'arrive_time': '2021-08-16 09:48:01', 'create_time': '2021-08-16 09:48:01', 'leave_time': '2021-08-16 09:48:01', 'region': 1122, 'grade': 1, 'province': '浙江'}礼物记录：live_gift_log（topic名）：
{ 'user_id': 3994 , 'gift_type': 3 , 'gift_total_amount': 28 , 'room_id': 20210813 , 'ip': '123.0.0.105' , 'create_time': '2021-08-16 09:46:51' , 'update_time': '2021-08-16 09:46:51'}模块记录表：live_module_roomid（Hbase维表）：
- Oceanus SQL作业编写
全网观看直播用户分布。（需提前在MySQL建表）
- 定义source：
CREATE TABLE `live_streaming_log_source ` ( `user_id` BIGINT, `ip` VARCHAR, `room_id` BIGINT, `arrive_time` TIMESTAMP, `leave_time` TIMESTAMP, `create_time` TIMESTAMP, `region_code` INT, `grade` INT, `province` VARCHAR ) WITH ( 'connector' = 'kafka', 'topic' = 'live_streaming_log', 'scan.startup.mode' = 'earliest-offset', 'properties.bootstrap.servers' = '172.28.28.13:9092', 'properties.group.id' = 'joylyu-consumer-2', 'format' = 'json', 'json.ignore-parse-errors' = 'false', 'json.fail-on-missing-field' = 'false' );- 定义sink：
CREATE TABLE `live_streaming_log_sink` ( `user_id` BIGINT, `ip` VARCHAR, `room_id` BIGINT, `arrive_time` TIMESTAMP, `leave_time` TIMESTAMP, `create_time` TIMESTAMP, `region_code` INT, `grade` INT, `province` VARCHAR, primary key(`user_id`, `ip`,`room_id`,`arrive_time`) not enforced) WITH ( 'connector' = 'jdbc', 'url' ='jdbc:mysql://172.28.28.227:3306/livedb?rewriteBatchedStatements=true&serverTimezon=Asia/Shanghai', 'table-name' = 'live_streaming_log', 'username' = 'root', 'password' = 'xxxxx', 'sink.buffer-flush.max-rows' = '5000', 'sink.buffer-flush.interval' = '2s', 'sink.max-retries' = '3');- 业务逻辑：
INSERT INTO `live_streaming_log_sink`SELECT `*` FROM `live_streaming_log_source`;- 礼物总和统计（需提前在MySQL建表）
- 定义source：
CREATE TABLE ` live_gift_total_source` ( `user_id` VARCHAR, `gift_type` VARCHAR, `gift_total_amount` BIGINT, `ip` VARCHAR, `create_time` VARCHAR) WITH ( 'connector' = 'kafka', 'topic' = 'live_gift_total', 'scan.startup.mode' = 'earliest-offset', 'properties.bootstrap.servers' = '172.28.28.13:9092', 'properties.group.id' = 'joylyu-consumer-1', 'format' = 'json', 'json.ignore-parse-errors' = 'false', 'json.fail-on-missing-field' = 'false' );- 定义sink：
CREATE TABLE `live_gift_total_sink` (`gift_type` VARCHAR,`gift_total_amount` BIGINT,primary key(`user_id`, `gift_type`) not enforced) WITH ('connector' = 'jdbc','url' = 'jdbc:mysql://172.28.28.227:3306/livedb?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai','table-name' = 'live_gift_total','username' = 'root','password' = 'xxxxx','sink.buffer-flush.max-rows' = '5000','sink.buffer-flush.interval' = '2s','sink.max-retries' = '3');- 业务逻辑：
INSERT INTO `live_gift_total_sink`SELECT `gift_type`, SUM(`gift_total_amount`) as `gift_total_amount_all`FROM `live_gift_total_source`GROUP BY `gift_type`;- 各模块礼物统计（需提前在MySQL建表）
- 定义source：
CREATE TABLE `live_gift_total_source` (`user_id` VARCHAR,`gift_type` VARCHAR,`gift_total_amount` BIGINT,`ip` VARCHAR,`create_time` VARCHAR,proc_time AS PROCTIME()) WITH ( 'connector' = 'kafka', 'topic' = 'live_gift_total', 'scan.startup.mode' = 'earliest-offset', 'properties.bootstrap.servers' = '172.28.28.13:9092', 'properties.group.id' = 'joylyu-consumer-1', 'format' = 'json', 'json.ignore-parse-errors' = 'false', 'json.fail-on-missing-field' = 'false' );- 定义Hbase维表：
CREATE TABLE `dim_hbase` (`rowkey` STRING,`cf` ROW <`module_id` STRING>,PRIMARY KEY (`rowkey`) NOT ENFORCED) WITH ('connector' = 'hbase-1.4','table-name' = 'dim_hbase','zookeeper.quorum' = '用户自己的hbase服务器zookeeper地址');- 定义sink：
CREATE TABLE `module_gift_total_sink` (`module_id` BIGINT,`module_gift_total_amount` BIGINT,primary key(`module_id`) not enforced) WITH ('connector' = 'jdbc','url' = 'jdbc:mysql://172.28.28.227:3306/livedb?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai','table-name' = 'live_gift_total','username' = 'root','password' = 'xxxxx','sink.buffer-flush.max-rows' = '5000','sink.buffer-flush.interval' = '2s','sink.max-retries' = '3');- 业务逻辑：
INSERT INTO `module_gift_total_sink`SELECT`b`.`cf`.`module_id`,SUM(`a`.`gift_total_amount`) AS `module_gift_total_amount`FROM `live_gift_total_source` AS `a`LEFT JOIN `dim_hbase` AS `b` for SYSTEM_TIME as of `a`.`proc_time` ON `a`.`room_id` = `b`.`rowkey`GROUP BY `b`.`cf`.`module_id`;（二）实时大屏可视化展示
- 添加数据源
进入商业智能分析界面，点击添加数据源->MySQL，按上面方法连接到指定MySQL数据库，点击保存。
- 创建数据集
点击创建数据集->SQL数据集（可根据实际业务场景选择其他数据集），从刚才的数据源中添加数据集，点击保存。
- 制作报告
新建报告。点击制作报告->新建报告（可选择任意模版），拖拽组件到中间空白处完成报告的制作。
设置实时刷新。点击左上角报告设置->高级，勾选获取实时数据，刷新间隔设置为3s（根据实际业务情况自行选择），这样可以根据MysQL数据源间隔3s一次自动刷新报告。完成之后点击保存。具体步骤见：
（https://cloud.tencent.com/document/product/590/19753）
- 查看报告
点击查看报告，选择刚才保存的报告，可以动态展示报告。注：此报告只做演示使用，可以参考:
（https://cloud.tencent.com/document/product/590/19784）
如下图所示，大屏中总共6个图表。
图表1：用户地区分布。表示观看直播客户在全国范围内的地区分布；
图表2：各级别会员人数。表示各个会员等级的总人数；
图表3：礼物类型总和。表示收到各礼物类型的总和；
图表4：最近6h礼物总数统计。表示最近6小时收到的礼物总计和；
图表5：刷礼物排行前10。表示刷礼物最多的10个客户；
图表6：在线人数。当天每个时间段进入直播间的人数。
实时大屏
四、总结
通过腾讯云CKafka组件采集数据，在兼容Flink开源版本的流计算Oceanus中实时进行维表关联等加工处理，将加工后的数据存储在MySQL等数据库中，最终通过商业智能分析BI组件实时刷新MySQL的数据绘制出了实时大屏，得到了实时刷新的效果。这个方案在数据库表设计时为了简便易懂做了简化处理，重点打通腾讯云产品展现整个方案。限于个人水平，如有理解有误之处欢迎批评指正。
（作者：spiderwu，腾讯CSIG高级工程师）
点击「阅读原文」，了解腾讯云流计算Oceanus更多信息~
