ksqlDB复杂处理场景：按时间窗口统计商品销量与营收

针对你遇到的窗口表与非窗口流关联报错、全局表数据量顾虑的问题，提供两种更优雅的处理方案：

方案一：流内直接关联+窗口聚合（无额外表存储） #

如果商品信息（articles）随订单一起传递且无需独立维护，可直接在流处理阶段完成订单行与对应商品的关联，再做窗口聚合：

1. 创建原始订单流（可根据实际数据结构调整）：

CREATE STREAM receipts (
    order_id STRING,
    order_time TIMESTAMP,
    articles ARRAY<STRUCT<id STRING, name STRING>>,
    line_items ARRAY<STRUCT<reference_id STRING, quantity INT, unit_price DECIMAL(10,2)>>
) WITH (
    KAFKA_TOPIC='receipts_topic',
    VALUE_FORMAT='JSON',
    KEY_FORMAT='KAFKA',
    TIMESTAMP='order_time'
);

2. 拆分订单行流：

CREATE STREAM receipt_line_items AS
SELECT
    order_id,
    order_time,
    li.reference_id AS article_id,
    li.quantity,
    li.unit_price,
    articles
FROM receipts
LATERAL VIEW EXPLODE(line_items) AS li
EMIT CHANGES;

3. 关联订单行与对应商品：

CREATE STREAM receipt_article_line_items AS
SELECT
    rli.order_id,
    rli.order_time,
    rli.article_id,
    a.name AS article_name,
    rli.quantity,
    rli.unit_price,
    rli.quantity * rli.unit_price AS revenue
FROM receipt_line_items rli
LATERAL VIEW EXPLODE(rli.articles) AS a
WHERE a.id = rli.article_id
EMIT CHANGES;

4. 1小时滚动窗口聚合统计：

CREATE TABLE article_sales_stats AS
SELECT
    article_id,
    article_name,
    SUM(quantity) AS total_sales,
    SUM(revenue) AS total_revenue,
    WINDOWSTART AS window_start,
    WINDOWEND AS window_end
FROM receipt_article_line_items
WINDOW TUMBLING (SIZE 1 HOUR, RETENTION 24 HOUR) -- 窗口保留24小时，过期自动清理
GROUP BY article_id, article_name
EMIT CHANGES;

方案二：带TTL的全局商品表+流关联+窗口聚合（适合商品信息需独立维护） #

如果商品信息可能单独更新，可将商品数据提取为带TTL（自动过期清理）的全局表，避免数据无限累积：

1. 创建带TTL的全局商品表：

CREATE TABLE articles_global AS
SELECT
    a.id AS article_id,
    LATEST_BY_OFFSET(a.name) AS article_name
FROM receipts
LATERAL VIEW EXPLODE(articles) AS a
GROUP BY a.id
WITH (TTL='30 DAYS'); -- 设置30天TTL，自动清理长期未更新的商品数据

2. 拆分订单行流：

CREATE STREAM receipt_line_items AS
SELECT
    order_time,
    li.reference_id AS article_id,
    li.quantity,
    li.unit_price,
    li.quantity * li.unit_price AS revenue
FROM receipts
LATERAL VIEW EXPLODE(line_items) AS li
EMIT CHANGES;

3. 订单行流关联全局商品表：

CREATE STREAM receipt_line_items_with_article AS
SELECT
    rli.order_time,
    rli.article_id,
    ag.article_name,
    rli.quantity,
    rli.revenue
FROM receipt_line_items rli
LEFT JOIN articles_global ag ON rli.article_id = ag.article_id
EMIT CHANGES;

4. 1小时滚动窗口聚合统计：

CREATE TABLE article_sales_stats AS
SELECT
    article_id,
    article_name,
    SUM(quantity) AS total_sales,
    SUM(revenue) AS total_revenue,
    WINDOWSTART AS window_start,
    WINDOWEND AS window_end
FROM receipt_line_items_with_article
WINDOW TUMBLING (SIZE 1 HOUR, RETENTION 24 HOUR)
GROUP BY article_id, article_name
EMIT CHANGES;

两种方案都避免了窗口表与非窗口流的关联冲突，且通过窗口RETENTION或表TTL控制数据量，无需担心存储压力。

内容的提问来源于stack exchange，提问作者filpa