本方案结合腾讯云 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]
注意私有网络和子网选择之前创建的网络和子网
Kafka集群
创建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)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’云数据库 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; --创建数据库列表
商业智能分析（Business Intelligence，BI）支持自服务数据准备、探索式分析和企业级管控，是新一代的敏捷自助型 BI 服务平台。只需几分钟，您就可以在云端轻松自如地完成数据分析、业务数据探查、报表制作等一系列数据可视化操作。便捷的拖拉拽式交互操作方式，让您无需依赖 IT 人员，无需担心试错成本，快速洞察数据背后的关联、趋势和逻辑。
页面地址：[https://cloud.tencent.com/product/bi]
BI购买
2 . 子用户提出申请
子用户申请
3 . 主账号审核通过。并给子用户授予添加数据源，创建数据集，查看报告的权限。
（这里选用开启外网方式连接，更多连接方式见官方文档：[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，填写完成后点击测试连接。
接下来通过案例为您介绍如何利用流计算服务 Oceanus 实现视频直播数字化运营的实时可视化数据处理与分析。
这里只列取以下 3 种统计指标：
事件 log：live_streaming_log（topic）
{
'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 维表）
全网观看直播用户分布（需提前在 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：在线人数。当天每个时间段进入直播间的人数。
实时大屏
通过腾讯云CKafka组件采集数据，在兼容Flink开源版本的流计算Oceanus中实时进行维表关联等加工处理，将加工后的数据存储在MySQL等数据库中，最终通过商业智能分析BI组件实时刷新MySQL的数据绘制出了实时大屏，得到了实时刷新的效果。这个方案在数据库表设计时为了简便易懂做了简化处理，重点打通腾讯云产品展现整个方案。限于个人水平，如有理解有误之处欢迎批评指正。
本文由哈喽比特于3年以前收录，如有侵权请联系我们。
文章来源：https://mp.weixin.qq.com/s/LC0ILUE1I8ts8pyzfZf6zA
京东创始人刘强东和其妻子章泽天最近成为了互联网舆论关注的焦点。有关他们“移民美国”和在美国购买豪宅的传言在互联网上广泛传播。然而，京东官方通过微博发言人发布的消息澄清了这些传言，称这些言论纯属虚假信息和蓄意捏造。
日前，据博主“@超能数码君老周”爆料，国内三大运营商中国移动、中国电信和中国联通预计将集体采购百万台规模的华为Mate60系列手机。
据报道，荷兰半导体设备公司ASML正看到美国对华遏制政策的负面影响。阿斯麦（ASML）CEO彼得·温宁克在一档电视节目中分享了他对中国大陆问题以及该公司面临的出口管制和保护主义的看法。彼得曾在多个场合表达了他对出口管制以及中荷经济关系的担忧。
今年早些时候，抖音悄然上线了一款名为“青桃”的 App，Slogan 为“看见你的热爱”，根据应用介绍可知，“青桃”是一个属于年轻人的兴趣知识视频平台，由抖音官方出品的中长视频关联版本，整体风格有些类似B站。
日前，威马汽车首席数据官梅松林转发了一份“世界各国地区拥车率排行榜”，同时，他发文表示：中国汽车普及率低于非洲国家尼日利亚，每百户家庭仅17户有车。意大利世界排名第一，每十户中九户有车。
近日，一项新的研究发现，维生素 C 和 E 等抗氧化剂会激活一种机制，刺激癌症肿瘤中新血管的生长，帮助它们生长和扩散。
据媒体援引消息人士报道，苹果公司正在测试使用3D打印技术来生产其智能手表的钢质底盘。消息传出后，3D系统一度大涨超10%，不过截至周三收盘，该股涨幅回落至2%以内。
9月2日，坐拥千万粉丝的网红主播“秀才”账号被封禁，在社交媒体平台上引发热议。平台相关负责人表示，“秀才”账号违反平台相关规定，已封禁。据知情人士透露，秀才近期被举报存在违法行为，这可能是他被封禁的部分原因。据悉，“秀才”年龄39岁，是安徽省亳州市蒙城县人，抖音网红，粉丝数量超1200万。他曾被称为“中老年...
9月3日消息，亚马逊的一些股东，包括持有该公司股票的一家养老基金，日前对亚马逊、其创始人贝索斯和其董事会提起诉讼，指控他们在为 Project Kuiper 卫星星座项目购买发射服务时“违反了信义义务”。
据消息，为推广自家应用，苹果现推出了一个名为“Apps by Apple”的网站，展示了苹果为旗下产品（如 iPhone、iPad、Apple Watch、Mac 和 Apple TV）开发的各种应用程序。
特斯拉本周在美国大幅下调Model S和X售价，引发了该公司一些最坚定支持者的不满。知名特斯拉多头、未来基金（Future Fund）管理合伙人加里·布莱克发帖称，降价是一种“短期麻醉剂”，会让潜在客户等待进一步降价。
据外媒9月2日报道，荷兰半导体设备制造商阿斯麦称，尽管荷兰政府颁布的半导体设备出口管制新规9月正式生效，但该公司已获得在2023年底以前向中国运送受限制芯片制造机器的许可。
近日，根据美国证券交易委员会的文件显示，苹果卫星服务提供商 Globalstar 近期向马斯克旗下的 SpaceX 支付 6400 万美元（约 4.65 亿元人民币）。用于在 2023-2025 年期间，发射卫星，进一步扩展苹果 iPhone 系列的 SOS 卫星服务。
据报道，马斯克旗下社交平台𝕏（推特）日前调整了隐私政策，允许 𝕏 使用用户发布的信息来训练其人工智能（AI）模型。新的隐私政策将于 9 月 29 日生效。新政策规定，𝕏可能会使用所收集到的平台信息和公开可用的信息，来帮助训练 𝕏 的机器学习或人工智能模型。
9月2日，荣耀CEO赵明在采访中谈及华为手机回归时表示，替老同事们高兴，觉得手机行业，由于华为的回归，让竞争充满了更多的可能性和更多的魅力，对行业来说也是件好事。
《自然》30日发表的一篇论文报道了一个名为Swift的人工智能（AI）系统，该系统驾驶无人机的能力可在真实世界中一对一冠军赛里战胜人类对手。
近日，非营利组织纽约真菌学会（NYMS）发出警告，表示亚马逊为代表的电商平台上，充斥着各种AI生成的蘑菇觅食科普书籍，其中存在诸多错误。
社交媒体平台𝕏（原推特）新隐私政策提到：“在您同意的情况下，我们可能出于安全、安保和身份识别目的收集和使用您的生物识别信息。”
2023年德国柏林消费电子展上，各大企业都带来了最新的理念和产品，而高端化、本土化的中国产品正在不断吸引欧洲等国际市场的目光。
罗永浩日前在直播中吐槽苹果即将推出的 iPhone 新品，具体内容为：“以我对我‘子公司’的了解，我认为 iPhone 15 跟 iPhone 14 不会有什么区别的，除了序（列）号变了，这个‘不要脸’的东西，这个‘臭厨子’。
