流式数据库 RisingWave「Demo」：直播指标实时分析

直播因其能与观众进行实时互动的独特优势，成为目前最为流行的娱乐形式之一。想要优化直播效果，有许多指标需要跟踪。最常见的指标比如：人流量、评论数量、直播卡顿时长等等。

本教程将分享如何使用 RisingWave 监控直播流量指标。我们为本教程设置了一个演示集群，以便大家可以轻松尝试。

1. 开始之前

确保您的环境中安装了 Docker 和 Docker Compose。请注意，Docker Compose 包含在 Windows 和 macOS 的 Docker Desktop 中。如果您使用 Docker Desktop，请确保在启动演示集群之前已经运行。
确保 PostgreSQL 交互式终端 psql 已安装在您的环境中。有关详细说明，请参阅下载 PostgreSQL。

2. 启动演示集群

在演示集群中，我们打包了 RisingWave 和一个工作负载生成器。一旦集群启动，工作负载生成器将开始生成随机流量并将它们发送到 Kafka。

首先，将 RisingWave 仓库克隆到环境中。

git clone https://github.com/risingwavelabs/risingwave.git

导航到 integration_tests/livestream 目录，并从 docker compose 文件启动演示集群。

cd risingwave/integration_tests/livestream
docker compose up -d

命令未找到？
-Compose V2 中的默认命令行句法以 docker compose 开头。详见 Docker 文档。
-如果您使用的是 Compose V1，请改用 docker-compose。

必要的 RisingWave 组件将被启动，包括 Frontend 节点、Compute 节点、Meta 节点和 MinIO。工作负载生成器将开始生成随机数据并将它们发送到 Kafka topic。在这个演示集群中，物化视图的数据将存储在 MinIO 实例中。

连接到 RisingWave 以管理数据流并执行数据分析。

psql -h localhost -p 4566 -d dev -U root

3. 将 RisingWave 连接到数据流

我们已经使用演示集群在 Kafka 中设置了数据流（以 JSON 格式），我们可以使用以下 SQL 语句连接到这些流。数据包含了直播流量指标的信息以及每个流的独特观众数量。

CREATE MATERIALIZED VIEW live_video_qos_10min AS
SELECT
    window_start AS report_ts,
    room_id,
    SUM(video_total_freeze_duration) AS video_total_freeze_duration,
    AVG(video_lost_pps) AS video_lost_pps,
    AVG(video_rtt) AS video_rtt
FROM
    TUMBLE(
        live_stream_metrics,
        report_timestamp,
        INTERVAL '10' MINUTE
    )
GROUP BY
    window_start,
    room_id;

RisingWave 已连接到流，但尚未开始消费数据。要处理数据，我们需要定义物化视图。创建物化视图后，RisingWave 将从指定的偏移量开始消费数据。

4. 创建物化视图

在本教程中，我们将创建不同的物化视图，用于跟踪直播流量性能和观众数量。

4.1 设置直播流量性能的物化视图

第一个物化视图将总结每个流每 10 分钟的流量性能。为了创建它，我们将使用 tumble 函数将每个事件映射到一个 10 分钟的窗口，并根据每个房间聚合以计算流卡顿的时长、每秒平均丢包数和平均往返时间。

CREATE MATERIALIZED VIEW live_video_qos_10min AS
SELECT
    window_start AS report_ts,
    room_id,
    SUM(video_total_freeze_duration) AS video_total_freeze_duration,
    AVG(video_lost_pps) AS video_lost_pps,
    AVG(video_rtt) AS video_rtt
FROM
    TUMBLE(
        live_stream_metrics,
        report_timestamp,
        INTERVAL '10' MINUTE
    )
GROUP BY
    window_start,
    room_id;

我们可以使用以下 SQL 语句查询结果。

SELECT * FROM live_video_qos_10min ORDER BY room_id, report_ts;

这里是一个示例结果。

report_ts	room_id	video_total_freeze_duration	video_lost_pps	video_rtt
2022-10-19 11:30:00+00:00	3998783950	1528	4.64286	196.21429
2022-10-19 11:40:00+00:00	3998783950	3635	4.66667	196.53333
2022-10-19 11:50:00+00:00	3998783950	603	4.09091	175.18182
2022-10-19 11:30:00+00:00	658508327	1431	4.32143	201.35714
2022-10-19 11:40:00+00:00	658508327	3619	5.23333	191.86667

4.2 设置观众数量的物化视图

接下来，我们将设置两个物化视图来跟踪观众数量。

第一个物化视图将每分钟跟踪整个直播网站的独特观众数量。我们将使用 tumble 函数将每个事件映射到一个一分钟的窗口，并计算每个时间窗口内的不同观众数量。

-- 实时总 UV 数据看板。
CREATE MATERIALIZED VIEW total_user_visit_1min AS
SELECT
    window_start AS report_ts,
    COUNT(DISTINCT user_id) AS uv
FROM
    TUMBLE(
        live_stream_metrics,
        report_timestamp,
        INTERVAL '1' MINUTE
    )
GROUP BY
    window_start;

我们可以使用以下 SQL 语句查询结果。

SELECT * FROM total_user_visit_1min ORDER BY report_ts;

这里是一个示例结果。

report_ts	uv
2022-10-19 11:35:00+00:00	2
2022-10-19 11:36:00+00:00	2
2022-10-19 11:37:00+00:00	2
2022-10-19 11:38:00+00:00	2
2022-10-19 11:39:00+00:00	2

第二个物化视图将每分钟跟踪每个主播的独特观众数量。我们将使用 tumble函数将每个事件映射到一个一分钟的窗口，然后按 room_id 分组，计算每个主播的独特观众数量。

CREATE MATERIALIZED VIEW room_user_visit_1min AS
SELECT
    window_start AS report_ts,
    COUNT(DISTINCT user_id) AS uv,
    room_id
FROM
    TUMBLE(
        live_stream_metrics,
        report_timestamp,
        INTERVAL '1' MINUTE
    )
GROUP BY
    window_start,
    room_id;

我们可以使用以下 SQL 语句查询结果。

SELECT * FROM room_user_visit_1min ORDER BY room_id, report_ts;

这里是一个示例结果。

report_ts	uv	room_id
2022-10-19 11:35:00+00:00	1	3998783950
2022-10-19 11:36:00+00:00	1	3998783950
2022-10-19 11:37:00+00:00	1	3998783950
2022-10-19 11:38:00+00:00	1	3998783950
2022-10-19 11:39:00+00:00	1	3998783950