实时广告效果分析

RisingWave 让我们能以低代码的方式进行实时广告效果分析。

概览

广告平台和广告商都希望衡量他们广告的表现。他们实现了一些事件，这些事件会在用户在网站或移动应用上与广告互动（如查看或点击广告、安装应用或购买商品）时触发并发送回他们的服务器。基于这些事件，他们定义了各种指标，从不同角度分析广告表现。

点击率（CTR）是数字广告中用来衡量广告效果的关键指标之一。它是点击次数除以展示次数的比值。展示次数是数字广告在应用或网站上显示在某人屏幕上的次数。高点击率意味着用户觉得展示给他们的广告有用且相关。

在本教程中，您将学习如何使用 RisingWave 从广告展示和点击事件中获取实时点击率。我们为这个教程专门设置了一个演示集群，以便您可以轻松尝试。

开始之前

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

第 1 步：启动演示集群

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

首先，将 risingwave 仓库克隆到您的环境中。

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

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

cd risingwave/integration_tests/ad-ctr
docker compose up -d

命令未找到？

Compose V2 中的默认命令行句法以 docker compose 开头。详见 Docker文档。

如果您正在使用 Compose V1，请改用 docker-compose。

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

第 2 步：将 RisingWave 连接到数据流

让我们连接到 RisingWave，以便我们可以管理数据流并执行数据分析。

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

我们将把广告展示和广告点击事件视为独立的流，并使用简化的 Schema，以便您可以轻松掌握要点。

以下是广告展示事件的 Schema。在此 Schema 中，impression_timestamp 是广告呈现给观众的日期和时间，bid_id 是在线广告的出价请求或活动的标识符。当我们计算点击率时，我们必须确保展示和点击是针对同一出价请求/活动的。否则，结果将毫无意义。

{
  "bid_id": 2439384144522347,
  "ad_id": 5,
  "impression_timestamp": "2022-05-23T14:11:04Z"
}

广告点击事件的schema如下：

{
  "bid_id": 2439384144522347,
  "click_timestamp": "2022-05-23T14:12:56Z"
}

对于相同的出价 ID，impression_timestamp 应始终小于（早于）click_timestamp。

我们已经使用演示集群在 Kafka 中以 JSON 格式设置了这两个数据流，我们可以使用以下 SQL 语句连接到这两个流。

CREATE SOURCE ad_impression (
    bid_id BIGINT,
    ad_id BIGINT,
    impression_timestamp TIMESTAMP WITH TIME ZONE
) WITH (
    connector = 'kafka',
    topic = 'ad_impression',
    properties.bootstrap.server = 'message_queue:29092',
    scan.startup.mode = 'earliest'
) FORMAT PLAIN ENCODE JSON;

CREATE SOURCE ad_click (
    bid_id BIGINT,
    click_timestamp TIMESTAMP WITH TIME ZONE
) WITH (
    connector = 'kafka',
    topic = 'ad_click',
    properties.bootstrap.server = 'message_queue:29092',
    scan.startup.mode = 'earliest'
) FORMAT PLAIN ENCODE JSON;

tip

scan.startup.mode = 'earliest' 意味着 Source 将从 Kafka 中最早的条目开始传输。RisingWave 内部会记录消费的偏移量在持久状态中，这样在故障恢复期间，它将从最后消费的偏移量恢复。

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

第 3 步：定义物化视图并查询结果

在本教程中，我们将创建两个物化视图，一个用于标准 CTR，另一个用于每 5 分钟的时间窗口 CTR。标准 CTR 的物化视图旨在以简化的方式显示计算，而时间窗口 CTR 旨在显示现实世界中的 CTR 计算。

为标准 CTR 设置物化视图

让我们先来看看标准 CTR 的物化视图。通过这个物化视图，我们分别计算展示和点击次数，根据 ad_id 将它们连接起来，并根据最新的展示和点击次数计算 CTR。

CREATE MATERIALIZED VIEW ad_ctr AS
SELECT
    ad_clicks.ad_id AS ad_id,
    ad_clicks.clicks_count :: NUMERIC / ad_impressions.impressions_count AS ctr
FROM
    (
        SELECT
            ad_impression.ad_id AS ad_id,
            COUNT(*) AS impressions_count
        FROM
            ad_impression
        GROUP BY
            ad_id
    ) AS ad_impressions
    JOIN (
        SELECT
            ai.ad_id,
            COUNT(*) AS clicks_count
        FROM
            ad_click AS ac
            LEFT JOIN ad_impression AS ai ON ac.bid_id = ai.bid_id
        GROUP BY
            ai.ad_id
    ) AS ad_clicks ON ad_impressions.ad_id = ad_clicks.ad_id;

然后，您可以通过查询您刚刚创建的物化视图来了解广告的表现：

SELECT * FROM ad_ctr WHERE ad_id = 9;

以下是一个示例结果。

 ad_id |              ctr
-------+--------------------------------
     9 | 0.9256055363321799307958477509

为 5 分钟窗口 CTR 设置物化视图

如果我们想要每 5 分钟的 CTR，事情就会变得有点复杂。我们需要使用 "tumble" 函数将流中的每个事件映射到一个 5 分钟的窗口。我们将创建一个物化视图 ad_ctr_5min 来计算时间窗口 CTR。这个视图将把展示事件分配到时间窗口中，并为每个广告在每个时间窗口中聚合展示次数。您可以根据需要将 5 分钟替换为适合您的任何时间窗口。

CREATE MATERIALIZED VIEW ad_ctr_5min AS
SELECT
    ac.ad_id AS ad_id,
    ac.clicks_count :: NUMERIC / ai.impressions_count AS ctr,
    ai.window_end AS window_end
FROM
    (
        SELECT
            ad_id,
            COUNT(*) AS impressions_count,
            window_end
        FROM
            TUMBLE(
                ad_impression,
                impression_timestamp,
                INTERVAL '5' MINUTE
            )
        GROUP BY
            ad_id,
            window_end
    ) AS ai
    JOIN (
        SELECT
            ai.ad_id,
            COUNT(*) AS clicks_count,
            ai.window_end AS window_end
        FROM
            TUMBLE(ad_click, click_timestamp, INTERVAL '5' MINUTE) AS ac
            INNER JOIN TUMBLE(
                ad_impression,
                impression_timestamp,
                INTERVAL '5' MINUTE
            ) AS ai ON ai.bid_id = ac.bid_id
            AND ai.window_end = ac.window_end
        GROUP BY
            ai.ad_id,
            ai.window_end
    ) AS ac ON ai.ad_id = ac.ad_id
    AND ai.window_end = ac.window_end;

tip

您可以轻松地在 ad_ctr_5min 之上构建一个 CTR 实时数据看板。CTR 值是动态变化的，给定窗口中的每个广告 CTR 可以绘制为折线图中的一个点。最终，我们能够分析 CTR 随时间的变化。

让我们看看结果。请注意，您的结果会有所不同，因为流中的数据是由工作负载生成器随机生成的。

SELECT * FROM ad_ctr_5min;

ad_id |              ctr               |        window_end
-------+--------------------------------+---------------------------
| 0.8823529411764705882352941176 | 2022-05-24 06:25:00+00:00
| 0.8793103448275862068965517241 | 2022-05-24 06:30:00+00:00
| 0.880597014925373134328358209  | 2022-05-24 06:35:00+00:00
| 0.8285714285714285714285714286 | 2022-05-24 06:40:00+00:00
| 0.3636363636363636363636363636 | 2022-05-24 06:25:00+00:00
| 0.4464285714285714285714285714 | 2022-05-24 06:30:00+00:00
| 0.5918367346938775510204081633 | 2022-05-24 06:35:00+00:00
| 0.5806451612903225806451612903 | 2022-05-24 06:40:00+00:00
| 0.0975609756097560975609756098 | 2022-05-24 06:30:00+00:00
| 0.0983606557377049180327868852 | 2022-05-24 06:35:00+00:00
| 0.0789473684210526315789473684 | 2022-05-24 06:40:00+00:00
| 0.1129032258064516129032258065 | 2022-05-24 06:45:00+00:00
| 0.4166666666666666666666666667 | 2022-05-24 06:25:00+00:00
| 0.2881355932203389830508474576 | 2022-05-24 06:30:00+00:00
| 0.3181818181818181818181818182 | 2022-05-24 06:35:00+00:00
| 0.3076923076923076923076923077 | 2022-05-24 06:40:00+00:00

您可以几分钟后重新运行查询，看看结果是否有更新。

当您完成时，运行以下命令以断开 RisingWave 的连接。

\d

可选：要删除容器和生成的数据，请使用以下命令。

docker compose down -v

总结

在本教程中，我们学到了：

如何连接两个 Source 。
如何使用 tumble 时间窗口函数获取时间窗口聚合结果。

概览​

开始之前​

第 1 步：启动演示集群​

第 2 步：将 RisingWave 连接到数据流​

第 3 步：定义物化视图并查询结果​

为标准 CTR 设置物化视图​

为 5 分钟窗口 CTR 设置物化视图​

总结​

概览