播面

在 Flink SQL 中，实现电商大屏实时 Top-N 的核心难点在于“当前1小时内”（滑动窗口）与“每秒更新”（滑动步长）的结合。

Flink SQL 提供了官方推荐的 Window Top-N 语法。为了优雅且高效地实现这个需求，我们需要分为三步：定义数据源、计算滑动窗口内的商品销售额、使用 ROW_NUMBER() OVER() 计算 Top-N。

以下是完整的生产级实现方案和深度优化建议。

一、 核心 SQL 实现

1. 定义带有 Watermark 的数据源

Top-N 依赖事件时间（Event Time）进行窗口计算，必须定义 Watermark。

CREATE TABLE source_orders (
    order_id STRING,
    product_id STRING,
    amount DECIMAL(10, 2),
    order_time TIMESTAMP(3),
    -- 假设最大延迟为 2 秒
    WATERMARK FOR order_time AS order_time - INTERVAL '2' SECOND 
) WITH (
    'connector' = 'kafka',
    -- 其他 Kafka 配置...
);

2. 定义滑动窗口聚合（HOP Window）

“当前1小时内，每秒更新”在 Flink SQL 中的语义是：窗口大小（Size）为 1 小时，滑动步长（Slide）为 1 秒 的 HOP 窗口。

CREATE VIEW windowed_sales AS
SELECT
    window_start,
    window_end,
    product_id,
    SUM(amount) AS total_sales
FROM TABLE(
    HOP(TABLE source_orders, DESCRIPTOR(order_time), INTERVAL '1' SECOND, INTERVAL '1' HOUR)
)
GROUP BY window_start, window_end, product_id;

注：这里每过 1 秒，就会有一个 1 小时跨度的窗口触发计算并输出，从而满足“每秒更新”的需求。

3. 使用 ROW_NUMBER() 计算 Top-N

这是最优雅的部分：Flink SQL 优化器能够识别特定的 ROW_NUMBER() OVER (...) WHERE rownum <= N 模式，并将其在底层转换为高效的 TopN 物理算子，而不是全局全量排序。

-- 最终输出到 Sink（如 Redis, MySQL, HBase）
CREATE TABLE sink_top10_dashboard (
    window_end TIMESTAMP(3),
    rn BIGINT,
    product_id STRING,
    total_sales DECIMAL(10, 2),
    PRIMARY KEY (window_end, rn) NOT ENFORCED -- 定义主键，支持 Upsert
) WITH (
    'connector' = 'jdbc', -- 或者 redis / upsert-kafka
    ...
);

INSERT INTO sink_top10_dashboard
SELECT 
    window_end, 
    rn, 
    product_id, 
    total_sales
FROM (
    SELECT 
        window_start,
        window_end,
        product_id,
        total_sales,
        -- 按窗口分区，按销售额倒序赋予排名
        ROW_NUMBER() OVER (
            PARTITION BY window_start, window_end 
            ORDER BY total_sales DESC
        ) AS rn
    FROM windowed_sales
)
WHERE rn <= 10; -- Flink 识别此条件，仅保留 Top 10 的状态

二、 为什么说这套写法“优雅”？（底层原理）

1. 状态极简（TopN 算子优化）：
   普通的 ORDER BY 需要在内存中保留所有数据进行全局排序。但 Flink 识别到 ROW_NUMBER() ... WHERE rn <= 10 时，底层会使用一个专门的 TopN 算子。该算子在 State 中只维护 10 条数据。当新数据到来时，只做比较和替换，内存开销极小。
2. Append-Only 特性：
   配合窗口函数（Windowing TVF），由于计算是基于窗口触发的，最终流入 Sink 的数据是 Append-Only 的（窗口触发时输出最终结果），避免了复杂的 Retraction（撤回）流，极大降低了对下游数据库（如 MySQL/Redis）的写入压力。

三、 ⚠️ 生产环境的“致命”性能陷阱与优化

虽然上面的 SQL 完美贴合了你的需求，但 INTERVAL '1' SECOND（步长） 与 INTERVAL '1' HOUR（窗口大小） 的组合在 Flink 中是一个典型的性能杀手（状态爆炸）。

原因：一条数据到来，会被同时分配到 3600 （1小时=3600秒）个重叠的窗口中。如果大促期间 TPS 是 10,000，每秒钟 Flink 需要处理 36,000,000 次窗口聚合状态更新！

优雅的生产级降级方案（强烈推荐）：

在实际电商大屏开发中，通常会做以下业务妥协或技术优化：

方案 A：调整滑动步长（业务妥协）
大屏的“实时”通常不需要严格到 1 秒。将滑动步长调整为 10 秒 或 1 分钟，状态膨胀率会骤降至 360 倍或 60 倍，极大减轻集群压力。

-- 改为 10 秒滑动一次
HOP(TABLE source, DESCRIPTOR(order_time), INTERVAL '10' SECOND, INTERVAL '1' HOUR)

方案 B：连续查询 + 状态 TTL（近似 1 小时）
如果不使用严格的窗口，而是纯靠连续查询计算“全天 Top-N”，再利用下游（如 Redis）的过期机制或者 Flink 的 State TTL 来清理老数据。但这种方式无法实现严格的“减去一小时前的销售额”。

方案 C：两阶段聚合优化（技术手段）
如果业务死活要求 1 秒步长，可以在 SQL 前面加一层 MiniBatch 优化，或者在 Flink 侧开启以下参数，以减少 State 的频繁写入：

-- 开启微批处理
table.exec.mini-batch.enabled: true
table.exec.mini-batch.allow-latency: 1s
table.exec.mini-batch.size: 5000

四、 总结

最优雅的 Flink SQL Top-N 范式固定为：
嵌套子查询包含 ROW_NUMBER() OVER (PARTITION BY 窗口 ORDER BY 指标 DESC) + 外层查询过滤 WHERE rn <= N。
只要确保按此格式编写，Flink 就能赋予其最优的执行计划。最后注意防范大窗口小步长带来的状态爆炸即可。