视频直播：基于流计算 Oceanus（Flink） 的实时大屏分析
1 解决方案描述
1.1 概述
本方案结合腾讯云 CKafka、流计算 Oceanus、私有网络 VPC、商业智能分析BI等，对视频直播行业数字化运营进行实时可视化分析。分析指标包含观看直播人员的地区分布、各级别会员统计、各模块打赏礼物情况、在线人数等。
具体可参考 商业智能分析 BI 五分钟入门
2.6.2添加MySQL数据源
（这里选用开启外网方式连接，更多连接方式见官方文档：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，填写完成后点击测试连接。
3 方案实现
接下来通过案例为您介绍如何利用流计算服务Oceanus实现视频直播数字化运营的实时可视化数据处理与分析。
3.1 解决方案
3.1.1 业务目标
这里只列取以下3种统计指标：
- 全站观看直播用户分布
- 礼物总和统计
- 各模块礼物统计3.1.2 源数据格式
事件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 |
所属直播模块 |
3.1.3 Oceanus SQL作业编写
全网观看直播用户分布（需提前在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_amount_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_amount',
'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;3.2 实时大屏可视化展示
创建项目
开通服务后，进入 商业智能分析 BI 控制台，单击创建项目开始创建项目。在弹层中输入项目名称，选择项目 icon 颜色，单击创建并进入刚创建的项目。
添加数据源
数据源用于连接客户本地数据库或云数据库，是数据表加工的基础；数据表是对数据源进行加工后的结果，用于报表创作时使用。 进入之前创建的项目，点击左上方【数据】> 【数据源】> 【新建数据源】即可快速创建本地数据源或云上数据源。
创建页面
点击左上方【页面】>【创建页面】。编辑好后即可单击右上角【保存】>【发布】。
具体操作步骤可参见 商业智能分析 BI 五分钟入门。
查看页面
单击【看板】，选择刚才保存的报告，如下图所示，大屏中总共5个图表。
此处仅做展示使用，具体展示效果可根据业务情况做相应调整
- 图表1：最近6h礼物总数统计。表示最近6小时收到的礼物总计和。
- 图表2：各级别会员人数。表示各个会员等级的总人数。
- 图表3：礼物类型总和。表示收到各礼物类型的总和。
- 图表4：各地域会员等级人数。表示各个地域不同会员等级的人数
- 图表5：礼物 Top10 客户。表示刷礼物最多的10个客户。
