请教一个 Flink SQL 的问题，解决了星巴克感谢

我是一个 Flink 小白，最近有一个监控需求，想使用 Flink SQL 实现，但很多概念还没搞清楚，遇到一个问题卡壳了，在论坛里寻 Flink 大佬指点一二，解决了送一杯星巴克作为感谢！

Flink SQL 官网用客户(customer)和订单(order)举例，但都每分钟统计流表每个客户订单的数量。我的需求是每分钟统计维表全量每个客户订单的数量，也就是就算这一分钟某个客户没有下单，也需要统计一个 0 出来，用于做监控报警。

为了不暴露业务需求，调整为客户和订单的场景，如果有不恰当的地方还请指出，我再补充，SQL 如下：

CREATE TEMPORARY TABLE customers (
    id INT,
    name STRING
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://....'
);

CREATE TEMPORARY TABLE orders (
    order_id     STRING,
    customer_id  INT,
    order_time   TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '15' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = '...'
);

CREATE TEMPORARY VIEW order_per_minute AS
SELECT
    customer_id,
    count(*) as cnt,
    TUMBLE_END(order_time, INTERVAL '1' MINUTE) AS window_end
FROM orders
GROUP BY customer_id, TUMBLE(tstamp, INTERVAL '1' MINUTE);

INSERT INTO destination
SELECT
    COALESCE(window_end, CURRENT_TIMESTAMP),
    customer_id,
    COALESCE(cnt, 0),
FROM
    customers LEFT JOIN order_per_minute
        ON customers.id = order_per_minute.customer_id;

实际执行上面的代码有问题，比如说有 3 个客户 c1/c2/c3 ，但只有 2 个客户 c1/c2 每分钟都下单，第一次执行结果是对的：

10:01, c1, 19  
10:01, c2, 32  
10:01, c3, 0

随后每分钟的数据，就会少掉 c3 的结果：

10:02, c1, 18  
10:02, c2, 22 // c3 没有输出
10:03, c1, 18  
10:03, c2, 22 // c3 没有输出

我也不清楚 Flink SQL 能否这么用吗，还是得用 DataStream API 解决？请论坛的 Flink 大佬帮忙看一下，感谢！

t3zb2xzvjm4yvmn

331 天前

实现过类似的需求，首先 lookup join 肯定是不行了，事实流中没有出现的用户肯定关联不到的。

比较彻底的解决方案是使用 datastream API ，process function 。还要状态编程，因为需要将维表的状态自己维护，比如把状态放到一个 tuple2 里，t0 是 customer id ，t1 给默认值 0 ，然后拿另外一个流的 element ，每来一条就给 t1+1 ，窗口触发时把所有的 tuple2 向下游发送。

需要额外考虑的一点是，有可能某个时间窗口内 1 个下单的都没有(比如半夜)，那么该窗口无法触发，没有任何输出，所以默认的滚动窗口\事件时间语义就不太行。还需要再实现一个窗口，事件时间、处理时间混合语义，保证即使没有事件仍然可以触发，输出所有用户下单数均为 0 的情况。

Flink SQL 没有实现过，但是有一个简单粗暴的想法，定时把维表的数据全量发到 flink ，构造出所有包含用户的事实流，这样你只需要改造 customers 维表那里就够了。
缺点是需要不断地读维表，对 MySQL 增加压力； source 端不断地向下游发维度信息，实际上不符合事件驱动和流式计算的原则。

我猜测这个需求数据量不大，实时性要求也没那么高，使用 spark streaming 可能是更好的选择。