视频直播:实时数据可视化分析

Posted 腾讯技术工程

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了视频直播:实时数据可视化分析相关的知识,希望对你有一定的参考价值。


作者:spiderwu,腾讯 CSIG 高级工程师

1 解决方案描述

1.1 概述

本方案结合腾讯云 CKafka、流计算 Oceanus、私有网络 VPC、商业智能分析 BI 等,对视频直播行业数字化运营进行实时可视化分析。分析指标包含观看直播人员的地区分布、各级别会员统计、各模块打赏礼物情况、在线人数等。

视频直播场景

1.2 方案架构及优势

根据以上视频直播场景,设计了如下架构图:

架构图

涉及产品列表:

  • 流计算 Oceanus

  • 私有网络 VPC

  • 消息队列 CKafka

  • 云数据库 mysql

  • EMR 集群 HBase 组件

  • 商业智能分析服务

2 前置准备

购买并创建相应的大数据组件。

2.1 创建 VPC 私有网络

私有网络是一块您在腾讯云上自定义的逻辑隔离网络空间,在构建 MySQL、EMR,ClickHouse 集群等服务时选择的网络必须保持一致,网络才能互通。否则需要使用对等连接、VPN 等方式打通网络。页面地址:https://console.cloud.tencent.com/vpc/vpc?rid=8

2.2 创建 Oceanus 集群

流计算 Oceanus 服务兼容原生的 Flink 任务。在 Oceanus 控制台的【集群管理】->【新建集群】页面创建集群,选择地域、可用区、VPC、日志、存储,设置初始密码等。VPC 及子网使用刚刚创建好的网络。创建完后 Flink 的集群如下:

Oceanus集群

2.3 创建消息队列 Ckafka

消息队列 CKafka(Cloud Kafka)是基于开源 Apache Kafka 消息队列引擎,提供高吞吐性能、高可扩展性的消息队列服务。消息队列 CKafka 完美兼容 Apache kafka 0.9、0.10、1.1、2.4、2.8 版本接口,在性能、扩展性、业务安全保障、运维等方面具有超强优势,让您在享受低成本、超强功能的同时,免除繁琐运维工作。页面地址:https://cloud.tencent.com/product/ckafka

2.2.1 创建 Ckafka 集群

注意私有网络和子网选择之前创建的网络和子网

Kafka集群
2.2.2 创建 topic



创建topic
2.2.3 模拟发送数据到 topic

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)

2.4 创建 EMR 集群

EMR 是云端托管的弹性开源泛 Hadoop 服务,支持 Spark、HBase、Presto、Flink、Druid 等大数据框架,本次示例主要需要使用 Flume、Hive、YARN、HUE、Oozie 组件。页面地址https://console.cloud.tencent.com/emr

1)在 EMR 集群中安装 HBase 组件。

HBase组件

2)如果生产环境,服务器配置可根据实际情况选择,示例中选择了低配服务器,网络需要选择之前创建好的 VPC 网络,始终保持服务组件在同一 VPC 下。

网络选择

3)进入 HBase Master 节点

HBaseMaster节点

4)点击登录进入服务器

5)创建 Hbase 表

# 进入HBase命令
[root@172~]# hbase shell
# 建表语句
create ‘dim_hbase’, ‘cf’

2.5 创建云数据库 MySQL

云数据库 MySQL(TencentDB for MySQL)是腾讯云基于开源数据库 MySQL 专业打造的高性能分布式数据存储服务,让用户能够在云中更轻松地设置、操作和扩展关系数据库。页面地址:https://console.cloud.tencent.com/cdb

新建 MySQL 服务的页面需要注意选择的网络是之前创建好的。

MySQL创建

创建完 MySQL 服务后,需要修改 binlog 参数,如图修改为 FULL(默认值为 MINIMAL)

mysql修改参数

修改完参数后,登陆 MySQL 创建示例所需要的数据库和数据库表。

1) 登陆 MySQL 云数据库

mysql登录

2) 新建数据库

打开 SQL 窗口或可视化页面创建数据库和表

CREATE DATABASE livedb;     --创建数据库列表

2.6 创建商业智能分析

商业智能分析(Business Intelligence,BI)支持自服务数据准备、探索式分析和企业级管控,是新一代的敏捷自助型 BI 服务平台。只需几分钟,您就可以在云端轻松自如地完成数据分析、业务数据探查、报表制作等一系列数据可视化操作。便捷的拖拉拽式交互操作方式,让您无需依赖 IT 人员,无需担心试错成本,快速洞察数据背后的关联、趋势和逻辑。

页面地址:https://cloud.tencent.com/product/bi

2.6.1 购买商业智能分析
  1. 需要主账号购买资源,购买时需根据创建的子账号数来进行购买。

BI购买
  1. 子用户提出申请

子用户申请
  1. 主账号审核通过。并给子用户授予添加数据源,创建数据集,查看报告的权限。

2.6.2 添加 MySQL 数据源

(这里选用开启外网方式连接,更多连接方式见官方文档:https://cloud.tencent.com/document/product/590/19294

1) 打开购买的 MySQL 实例,开启外网

mysql开启外网

2)将 SaaS BI(119.29.66.144:3306)添加到 MySQL 数据库安全组


添加安全组1

添加安全组2

注意添加的是 MySQL 3306 端口,不是外网映射的端口。


添加安全组3

3)创建 MySQL 账户并配置权限

创建账户,并设置账号密码,注意主机 IP 设置为%

创建账户1

创建账户2

设置账号权限:

设置权限1

设置权限2

4) 进入智能商业分析,连接 MySQL 数据库。添加数据源->MySQL,填写完成后点击测试连接。

3 方案实现

接下来通过案例为您介绍如何利用流计算服务 Oceanus 实现视频直播数字化运营的实时可视化数据处理与分析。

3.1 解决方案

3.1.1 业务目标

这里只列取以下 3 种统计指标:

  • 全站观看直播用户分布

  • 礼物总和统计

  • 各模块礼物统计 3.1.2 源数据格式

事件 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 维表)

3.1.2Oceanus 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_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`;

3.2 实时大屏可视化展示

3.2.1 添加数据源

进入商业智能分析界面,点击添加数据源->MySQL,按上面方法连接到指定 MySQL 数据库,点击保存。

3.2.2 创建数据集

点击创建数据集->SQL 数据集(可根据实际业务场景选择其他数据集),从刚才的数据源中添加数据集,点击保存。

3.2.3 制作报告

新建报告。点击制作报告->新建报告(可选择任意模版),拖拽组件到中间空白处完成报告的制作。

设置实时刷新。点击左上角报告设置->高级,勾选获取实时数据,刷新间隔设置为 3s(根据实际业务情况自行选择),这样可以根据 Mysq 数据源间隔 3s 一次自动刷新报告。完成之后点击保存。

具体步骤见官网文档:https://cloud.tencent.com/document/product/590/19753

3.2.4 查看报告

点击查看报告,选择刚才保存的报告,可以动态展示报告。(注:此报告只做演示使用,可以参考官方文档优化报告:https://cloud.tencent.com/document/product/590/19784

如下图所示,大屏中总共 6 个图表。

图表 1:用户地区分布。表示观看直播客户在全国范围内的地区分布;

图表 2:各级别会员人数。表示各个会员等级的总人数;

图表 3:礼物类型总和。表示收到各礼物类型的总和;

图表 4:最近 6h 礼物总数统计。表示最近 6 小时收到的礼物总计和;

图表 5:刷礼物排行前 10。表示刷礼物最多的 10 个客户;

图表 6:在线人数。当天每个时间段进入直播间的人数。

实时大屏

4 总结

通过腾讯云CKafka组件采集数据,在兼容Flink开源版本的流计算Oceanus中实时进行维表关联等加工处理,将加工后的数据存储在MySQL等数据库中,最终通过商业智能分析BI组件实时刷新MySQL的数据绘制出了实时大屏,得到了实时刷新的效果。这个方案在数据库表设计时为了简便易懂做了简化处理,重点打通腾讯云产品展现整个方案。限于个人水平,如有理解有误之处欢迎批评指正。

以上是关于视频直播:实时数据可视化分析的主要内容,如果未能解决你的问题,请参考以下文章

一小时搭建实时数据分析平台

LBMS直播数据可视化系统展示

34源码数据可视化:基于 Echarts + Python 动态实时大屏 - 视频平台

如何在三维可视化平台中接入实时的安防视频

R语言基础1:R语言开启数据分析之路(附视频)|暑期系列

如何在三维可视化平台中接入实时的安防视频