Flink 代码这么写，窗口能触发才怪（不良编程习惯）_flinksql怎么保证窗口一定触发

1.序篇-先说结论

本文主要记录小伙伴萌在使用 DataStream API 实现事件时间窗口类应用时会遇到的窗口不触发问题的坑以及其排查过程。

博主希望你在看完本文后一定要养成这个编程习惯：使用 DataStream API 实现 Flink 任务时，Watermark Assigner 能靠近 Source 节点就靠近 Source 节点，尽量前置。

要想问为啥，接着往下看！！！

我从以下几个章节说明上述的问题以及为什么这样建议，希望能抛砖引玉，带给大家一些启发。

1. ⭐ 踩坑场景篇-这个坑是啥样的

2. ⭐ 问题排查篇-坑的排查过程

3. ⭐ 问题原理解析篇-导致问题的机制是什么

4. ⭐ 避坑篇-如何避免这种问题

5. ⭐ 总结篇

2.踩坑场景篇-这个坑是啥样的

2.1.需求场景

首先介绍一下这个坑对应的一个需求场景以及第一版本的实现代码。

1. 需求：在电商平台中，需要根据网页在线用户的心跳日志（每 30s 上报一次用户心跳日志）计算当前这一分钟在购物车页面（Shopping-Cart）停留的在线人数。

2. 数据源：每 30s 上报一次的用户心跳日志（user_id、page、time 三个字段分别对应 用户 id、用户所在页面、日志上报时间）

3. 数据处理：先过滤出购物车按照时间戳对用户心跳日志进行滚动窗口（Tumble）聚合计算

4. 数据汇：每分钟聚合的结果数据（uv、time 两个字段分别对应 购物车页面的当前这一分钟的同时在线人数、当前这一分钟的时间戳）

Flink DataStream API 具体实现代码如下：

public class WatermarkTest {

    public static void main(String[] args) throws Exception {

        // 获取到 Flink 环境，博主自己封装的接口 FlinkEnv
        FlinkEnv flinkEnv = FlinkEnvUtils.getStreamTableEnv(args);

        // 设置并发度
        flinkEnv.env().setParallelism(100);

        flinkEnv.env()
                // 数据源：上报的日志
                .addSource(xxx)
                // 过滤出 购物车页面（Shopping-Cart）的数据
                .filter(new FilterFunction<SourceModel>() {
                    @Override
                    public boolean filter(SourceModel value) throws Exception {
                        return value.getPage().equals("Shopping-Cart");
                    }
                })
                // 分配 Watermark
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SourceModel>(Time.minutes(1)) {
                    @Override
                    public long extractTimestamp(SourceModel element) {
                        return element.getTime();
                    }
                })
                // 为了进行合并计算，shuffle 到一个算子中࿰
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27