📚 课程目录第3章 Window Join详解

3. Window Join 详解：Tumbling、Sliding、Session

双流 Join 里，Window Join 是最直观、最容易上手的一类。说白了，就是把两条流的数据，按照同一个时间窗口对齐，然后在窗口内做关联。我刚开始接触 Flink 时，最先搞明白的就是这个。

窗口类型有三种：滚动窗口（Tumbling）、滑动窗口（Sliding）、会话窗口（Session）。它们各自适用的场景完全不同。下面我一个个拆开讲，每个都会配上代码示例和我在项目中踩过的坑。

3.1 Tumbling Window Join：固定周期对齐

滚动窗口是最简单的。窗口大小固定，比如 10 秒。两条流的数据，只要时间戳落在同一个 10 秒桶里，就一起 Join。窗口之间不重叠，一条数据只属于一个窗口。

适用场景：每 5 分钟统计一次订单和支付的对账情况；每 1 小时计算一次广告曝光和点击的匹配率。说白了，就是按固定周期做快照式关联。

个人经验：我在做电商实时对账时，就用 Tumbling Window Join 把订单流和支付流按 1 分钟窗口对齐。当时踩了个坑——订单事件和支付事件的时间戳如果来自不同时钟源，偏差会很大。后来强制统一用订单创建时间作为事件时间，才把对账准确率提上去。

// 示例：Tumbling Window Join —— 订单流 <-> 支付流，每 10s 对齐
DataStream<OrderEvent> orderStream = ...;   // 订单事件流
DataStream<PayEvent> payStream = ...;       // 支付事件流

orderStream.join(payStream)
    .where(OrderEvent::getOrderId)
    .equalTo(PayEvent::getOrderId)
    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
    .apply(new JoinFunction<OrderEvent, PayEvent, String>() {
        @Override
        public String join(OrderEvent order, PayEvent pay) {
            return "Order " + order.getOrderId() + " paid at " + pay.getPayTime();
        }
    })
    .print();

提示：滚动窗口的 watermark 延迟建议设为窗口大小的 1/3 到 1/2。比如窗口 10s，延迟设 3-5s。太短容易丢数据，太长又增加延迟。我一般取 4s 作为默认值。

3.2 Sliding Window Join：滚动监控利器

滑动窗口有两个参数：窗口大小和滑动步长。窗口大小决定关联的时间范围，滑动步长决定计算频率。比如窗口 10s，滑动 5s，那每 5s 就会输出一次过去 10s 的 Join 结果。数据可能同时属于多个窗口。

适用场景：实时监控场景。比如每 30 秒统计一次过去 5 分钟的点击转化率；每 1 分钟计算一次过去 15 分钟的交易成功率。你想想看，这种场景下，数据重叠是必须的，否则监控曲线会断断续续。

注意：滑动窗口会产生数据放大。窗口大小 / 滑动步长 = 每个数据归属的窗口数。比如窗口 10s、滑动 1s，一条数据要 Join 10 次。我曾经在项目中没注意这个，结果状态暴涨，OOM 了。后来把滑动步长从 1s 改成 3s，压力立刻降下来。

// 示例：Sliding Window Join —— 点击流 <-> 转化流，每 30s 滑动一次
DataStream<ClickEvent> clickStream = ...;
DataStream<ConvertEvent> convertStream = ...;

clickStream.join(convertStream)
    .where(ClickEvent::getUserId)
    .equalTo(ConvertEvent::getUserId)
    .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.seconds(30)))
    .apply(new JoinFunction<ClickEvent, ConvertEvent, String>() {
        @Override
        public String join(ClickEvent click, ConvertEvent convert) {
            return "User " + click.getUserId() + " converted after click";
        }
    })
    .print();

参数	含义	建议
窗口大小	关联的时间范围	根据业务周期定，如 5min、15min
滑动步长	计算输出的频率	步长越小，数据放大越严重
数据放大倍数	窗口大小 / 滑动步长	建议控制在 10 倍以内

3.3 Session Window Join：自然活动段关联

会话窗口没有固定长度。它根据数据的活动间隙来划分窗口。你设置一个 session gap，比如 5 分钟。两条流的数据，如果时间戳相差不超过 5 分钟，就属于同一个会话窗口。一旦超过 5 分钟没有新数据，窗口就关闭。

适用场景：用户行为分析。比如把用户的浏览事件和加购事件按会话关联；把 App 启动事件和后续操作事件按自然活动段 Join。窗口长度不固定，完全由数据驱动。

避坑指南：我曾经在游戏行业做过一个项目，需要把玩家的登录事件和充值事件按会话关联。当时 gap 设了 30 分钟，结果发现有些玩家挂机时间很长，会话一直不关闭，状态越积越多。后来加了最大会话时长限制（比如 2 小时强制关闭），才把状态控制住。

// 示例：Session Window Join —— 浏览流 <-> 加购流，gap = 5min
DataStream<BrowseEvent> browseStream = ...;
DataStream<CartEvent> cartStream = ...;

browseStream.join(cartStream)
    .where(BrowseEvent::getSessionId)
    .equalTo(CartEvent::getSessionId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(5)))
    .apply(new JoinFunction<BrowseEvent, CartEvent, String>() {
        @Override
        public String join(BrowseEvent browse, CartEvent cart) {
            return "Session " + browse.getSessionId() + " has browse and cart";
        }
    })
    .print();

技巧：Session Window 的 gap 大小直接影响窗口数量和状态大小。gap 太小，会话碎片多；gap 太大，窗口合并频繁。我一般先用离线数据跑一遍，统计用户行为间隔的分布，再选一个合理的 gap 值。比如 80% 的用户操作间隔在 3 分钟以内，那 gap 设 5 分钟就够用了。

3.4 三种窗口的选择建议

嗯，这里我直接给一张对比表，方便你快速决策：

维度	Tumbling	Sliding	Session
窗口长度	固定	固定	不固定
窗口重叠	不重叠	重叠	不重叠
数据归属	一个窗口	多个窗口	一个窗口
状态压力	低	中（取决于滑动步长）	高（窗口合并频繁）
典型场景	固定周期对账	滚动指标监控	用户行为分析