本方案结合腾讯云CKafka、流计算Oceanus、私有网络VPC、商业智能分析BI等,对视频直播行业数字化运营进行实时可视化分析。分析指标包含观看直播人员的地区分布、各级别会员统计、各模块打赏礼物情况、在线人数等。
根据以上视频直播场景,设计了如下架构图:
涉及产品列表:
购买并创建相应的大数据组件。
私有网络是一块您在腾讯云上自定义的逻辑隔离网络空间,在构建MySQL、EMR,ClickHouse集群等服务时选择的网络必须保持一致,网络才能互通。否则需要使用对等连接、VPN等方式打通网络。页面地址:https://console.cloud.tencent.com/vpc/vpc?rid=8
流计算Oceanus服务兼容原生的Flink任务。在Oceanus控制台的【集群管理】->【新建集群】页面创建集群,选择地域、可用区、VPC、日志、存储,设置初始密码等。VPC及子网使用刚刚创建好的网络。创建完后Flink的集群如下:
消息队列 CKafka(Cloud Kafka)是基于开源 Apache Kafka 消息队列引擎,提供高吞吐性能、高可扩展性的消息队列服务。消息队列 CKafka 完美兼容 Apache kafka 0.9、0.10、1.1、2.4、2.8版本接口,在性能、扩展性、业务安全保障、运维等方面具有超强优势,让您在享受低成本、超强功能的同时,免除繁琐运维工作。页面地址:https://cloud.tencent.com/product/ckafka
注意私有网络和子网选择之前创建的网络和子网
1)kafka客户端
进入同子网的CVM下,启动kafka客户端,模拟发送数据,具体操作文档参考官网:
https://cloud.tencent.com/document/product/597/56840
2)使用脚本发送
脚本一:Java参考地址:https://cloud.tencent.com/document/product/597/54834
脚本二:Python脚本生成模拟数据:
#!/usr/bin/python3 # 首次使用该脚本,需 "pip3 install kafka" 安装kafka模块 import json import random import time from 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 = 0 pre_hour_count = 0 hour_unit = 3600 day_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是云端托管的弹性开源泛 Hadoop 服务,支持 Spark、HBase、Presto、Flink、Druid 等大数据框架,本次示例主要需要使用Flume、Hive、YARN、HUE、Oozie组件。页面地址https://console.cloud.tencent.com/emr
1)在EMR集群中安装HBase组件。
2)如果生产环境,服务器配置可根据实际情况选择,示例中选择了低配服务器,网络需要选择之前创建好的VPC网络,始终保持服务组件在同一VPC下。
3)进入HBase Master节点
4)点击登录进入服务器
5)创建Hbase表
# 进入HBase命令 [root@172~]# hbase shell # 建表语句 create ‘dim_hbase’, ‘cf’
云数据库 MySQL(TencentDB for MySQL)是腾讯云基于开源数据库 MySQL 专业打造的高性能分布式数据存储服务,让用户能够在云中更轻松地设置、操作和扩展关系数据库。页面地址:https://console.cloud.tencent.com/cdb
新建MySQL服务的页面需要注意选择的网络是之前创建好的。
创建完MySQL服务后,需要修改binlog参数,如图修改为FULL(默认值为MINIMAL)
修改完参数后,登陆MySQL创建示例所需要的数据库和数据库表。
1) 登陆MySQL云数据库
2) 新建数据库
打开SQL窗口或可视化页面创建数据库和表
CREATE DATABASE livedb; --创建数据库列表
商业智能分析(Business Intelligence,BI)支持自服务数据准备、探索式分析和企业级管控,是新一代的敏捷自助型 BI 服务平台。只需几分钟,您就可以在云端轻松自如地完成数据分析、业务数据探查、报表制作等一系列数据可视化操作。便捷的拖拉拽式交互操作方式,让您无需依赖 IT 人员,无需担心试错成本,快速洞察数据背后的关联、趋势和逻辑。
页面地址:https://cloud.tencent.com/product/bi
1) 需要主账号购买资源,购买时需根据创建的子账号数来进行购买。
2) 子用户提出申请
3) 主账号审核通过。并给子用户授予添加数据源,创建数据集,查看报告的权限。
(这里选用开启外网方式连接,更多连接方式见官方文档:https://cloud.tencent.com/document/product/590/19294)
1) 打开购买的MySQL实例,开启外网
2) 将SaaS BI(119.29.66.144:3306)添加到MySQL数据库安全组
注意添加的是MySQL 3306端口,不是外网映射的端口。
3) 创建MySQL账户并配置权限
创建账户,并设置账号密码,注意主机IP设置为%
设置账号权限,
4) 进入智能商业分析,连接MySQL数据库。添加数据源->MySQL,填写完成后点击测试连接。
接下来通过案例为您介绍如何利用流计算服务Oceanus实现视频直播数字化运营的实时可视化数据处理与分析。
这里只列取以下3种统计指标:
事件log:live_streaming_log(topic)
字段 | 类型 | 含义 |
---|---|---|
user_id | bigint | 客户号 |
ip | varchar | 客户ip地址 |
room_id | bigint | 房间号 |
arrive_time | timestamp | 进入房间时间 |
leave_time | timestamp | 离开房间时间 |
create_time | timestamp | 创建时间 |
region_code | int | 地区编码 |
grade | int | 会员等级 |
province | varchar | 所在省份 |
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 | bigint | 客户号 |
gift_type | int | 礼物类型 |
room_id | bigint | 房间号 |
gift_total_amount | int | 礼物数量 |
ip | varchar | ip地址 |
create_time | timestamp | 创建时间 |
{ '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维表)
字段 | 例子 | 含义 |
---|---|---|
room_id | 20210813 | 房间号 |
mudule_id | 1001 | 所属直播模块 |
全网观看直播用户分布(需提前在MySQL建表)
1、定义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' );
2、定义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' );
3、业务逻辑
INSERT INTO live\_streaming\_log\_sink SELECT \* FROM live\_streaming\_log\_source;
? 礼物总和统计(需提前在MySQL建表)
1、 定义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' );
2、 定义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' );
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建表)
1、 定义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' );
2、 定义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地址' );
3、 定义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' );
4、业务逻辑
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(根据实际业务情况自行选择),这样可以根据Mysq数据源间隔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:在线人数。当天每个时间段进入直播间的人数。
淘宝联盟网站推广必须备案吗?需要的。不管用哪里的服务器,要做淘宝联盟网站的...
您在进行 服务器租用 或者 服务器托管 之后,除非重装操作系统等特殊情况,都不需...
为了使组织保持相关性和竞争力,数字化转型日益重要。日益复杂的IT基础设施需要D...
unity中是自带合并网格功能的,并不用开发者遍历网格数据,然后再添加到一个列表中...
客户简介 中石化是中国最大的成品油和石化产品供应商、第二大油气生产商,是世界...
操作场景 用户可以在管理控制台修改专属主机的名称。 操作流程 登录管理控制台。...
精彩内容推荐 上海ACE同城会演讲实录|云原生分布式数据库PolarDB-X 在近期的上海...
数据融合需要标准规范先行,实现数据可见性、数据易理解性、数据可链接性、数据...
实时及未来,最近在腾讯云Oceanus进行实时计算服务,以下为mysql到flink到ES实践...
1.自从知道吃货总比痴货好后,我觉得还是当个吃货吧。 2.人常说不要让青春留白...