62.2 窗口与水位线

一句话

窗口决定「怎么切数据」，水位线决定「什么时候切完」；窗口类型是业务语义，水位线是正确性与延迟的权衡——两者混在一起调，最后一定说不清「为什么这个数少了」。

四种窗口

窗口	切法	一条事件属于几个窗口	状态成本	典型用途
滚动（tumbling）	固定长度、不重叠	1	低	每分钟 GMV、每小时订单量
滑动（sliding / hop）	固定长度、按步长滑动、可重叠	`size / slide`（近似）	中–高	5 分钟移动平均、最近 10 分钟失败率
会话（session）	按「活动间隔」合并，长度不固定	1	中（需合并）	用户行为会话、设备在线段
全局（global）	一个无限窗口，靠触发器切分	1	取决于触发器	自定义触发、全局累计

时间语义（事件时间/处理时间/摄入时间）的定义与取舍见 62.1 流处理模型与时间语义。

🔑 窗口不是「时间切分函数」这么简单：滚动窗口回答「这一分钟发生了什么」，滑动窗口回答「最近一段时间怎么样」，会话窗口回答「一次行为从哪开始到哪结束」。选错窗口，指标本身就没有业务含义。

水位线的三个要素

要素	做什么	生产上要盯什么
生成	从事件时间算 `W = max(event_time) - delay`	乱序分布、迟到比例、时钟漂移
传播	多输入算子取各输入水位的 min	空闲分区、慢分区、背压
触发	`window_end <= W` 时触发	触发频率、迟到更新、状态清理时机

水位线是单调递增的：一旦声明「≤ W 的数据到齐了」，就不能回退。它也是每个分区独立推进的；算子有多个输入时，输出水位取所有输入的最小值，否则快分区会把水位推过慢分区，慢分区的数据会被误判迟到。

⚠️ 一个空闲分区会让整个作业的水位停住。Flink 的 table.exec.source.idle-timeout 或 DataStream 的 withIdleness 就是把「长时间没有数据的分区」标记为空闲，让它不再拖住全局水位。

代码：三种延迟，三种完整度

事件时间与到达时间故意错开：A14 比 A11 先到，B3 迟到，A9 更晚。

package main

import (
	"fmt"
	"sort"
	"strings"
)

type Ev struct {
	ID   string
	User string
	Ts   int // 事件时间（秒）
	Arr  int // 到达时间（秒）
}

// 到达顺序和事件时间不一致：A14 比 A11 先到，B3 迟到，A9 更晚。
var events = []Ev{
	{"A2", "A", 2, 2},
	{"A4", "A", 4, 4},
	{"A14", "A", 14, 15},
	{"A11", "A", 11, 16},
	{"B3", "B", 3, 17},
	{"B12", "B", 12, 18},
	{"A19", "A", 19, 19},
	{"A9", "A", 9, 24},
	{"A21", "A", 21, 22},
	{"B22", "B", 22, 23},
	{"A33", "A", 33, 34},
	{"B35", "B", 35, 36},
}

const (
	size   = 10
	maxWin = 40
)

type result struct {
	counts   map[int]int
	dropped  []string
	emitAt   map[int]int
	maxDelay int
	accepted []Ev
}

func winStart(ts int) int { return ts / size * size }

func run(delay int) result {
	order := append([]Ev(nil), events...)
	sort.Slice(order, func(i, j int) bool { return order[i].Arr < order[j].Arr })

	r := result{counts: map[int]int{}, emitAt: map[int]int{}}
	closed := map[int]bool{}
	maxTS := -1

	closeUntil := func(wm, now int) {
		for s := 0; s < maxWin; s += size {
			if !closed[s] && s+size <= wm {
				closed[s] = true
				r.emitAt[s] = now
				if d := now - (s + size); d > r.maxDelay {
					r.maxDelay = d
				}
			}
		}
	}

	for _, e := range order {
		w := winStart(e.Ts)
		if closed[w] {
			r.dropped = append(r.dropped, e.ID)
			continue
		}
		r.counts[w]++
		r.accepted = append(r.accepted, e)
		if e.Ts > maxTS {
			maxTS = e.Ts
		}
		closeUntil(maxTS-delay, e.Arr)
	}
	// 流结束：把所有窗口按最终水位刷出
	for s := 0; s < maxWin; s += size {
		if !closed[s] {
			closed[s] = true
			r.emitAt[s] = order[len(order)-1].Arr
			if d := r.emitAt[s] - (s + size); d > r.maxDelay {
				r.maxDelay = d
			}
		}
	}
	return r
}

func main() {
	fmt.Println("=== 水位延迟对「完整度 vs 延迟」的影响（10s 滚动窗口）===")
	fmt.Printf("%-8s %-18s %-10s %-12s\n", "delay", "counts(0,10,20,30)", "dropped", "max-emit-delay")
	for _, d := range []int{0, 5, 20} {
		r := run(d)
		var cs []string
		for s := 0; s < maxWin; s += size {
			cs = append(cs, fmt.Sprint(r.counts[s]))
		}
		fmt.Printf("%-8d %-18s %-10d %ds\n", d, strings.Join(cs, ","), len(r.dropped), r.maxDelay)
	}

	r5 := run(5)
	fmt.Println("\n=== delay=5s 的窗口产出明细 ===")
	fmt.Printf("%-10s %-6s %-8s %-5s %s\n", "window", "count", "emit-at", "end", "events")
	for s := 0; s < maxWin; s += size {
		var ids []string
		for _, e := range r5.accepted {
			if winStart(e.Ts) == s {
				ids = append(ids, e.ID)
			}
		}
		fmt.Printf("%-10s %-6d %-8d %-5d %s\n", fmt.Sprintf("[%02d,%02d)", s, s+size), r5.counts[s], r5.emitAt[s], s+size, strings.Join(ids, ","))
	}
	fmt.Printf("迟到丢弃：%v\n", r5.dropped)

	fmt.Println("\n=== 滑动窗口（size=10s, slide=5s，按事件时间）===")
	fmt.Printf("%-14s %s\n", "window", "count")
	for s := 0; s < maxWin; s += 5 {
		c := 0
		for _, e := range r5.accepted {
			if e.Ts >= s && e.Ts < s+size {
				c++
			}
		}
		fmt.Printf("[%02d,%02d)      %d\n", s, s+size, c)
	}

	fmt.Println("\n=== 会话窗口（gap=5s，按用户，按事件时间）===")
	for _, user := range []string{"A", "B"} {
		var ts []int
		for _, e := range r5.accepted {
			if e.User == user {
				ts = append(ts, e.Ts)
			}
		}
		sort.Ints(ts)
		sessions := [][2]int{}
		for _, t := range ts {
			if len(sessions) == 0 || t-sessions[len(sessions)-1][1] > 5 {
				sessions = append(sessions, [2]int{t, t})
			} else {
				sessions[len(sessions)-1][1] = t
			}
		}
		fmt.Printf("user %s: %d 个会话 %v\n", user, len(sessions), sessions)
	}

	fmt.Println("\n结论：水位延迟 0 时 [0,10) 只有 2 条（B3 与 A9 被丢）；延迟 5s 补回 B3，但 A9 仍丢；")
	fmt.Println("      延迟 20s 全部收齐，代价是 [0,10) 的结果要等到 34s 才最终确定。")
}

本机实测（macOS，Apple M1，Go 1.27.1）：

=== 水位延迟对「完整度 vs 延迟」的影响（10s 滚动窗口）===
delay    counts(0,10,20,30) dropped    max-emit-delay
0        2,4,2,2            2          5s
5        3,4,2,2            1          14s
20       4,4,2,2            0          24s

=== delay=5s 的窗口产出明细 ===
window     count  emit-at  end   events
[00,10)    3      19       10    A2,A4,B3
[10,20)    4      34       20    A14,A11,B12,A19
[20,30)    2      36       30    A21,B22
[30,40)    2      36       40    A33,B35
迟到丢弃：[A9]

=== 滑动窗口（size=10s, slide=5s，按事件时间）===
window         count
[00,10)      3
[05,15)      3
[10,20)      4
[15,25)      3
[20,30)      2
[25,35)      1
[30,40)      2
[35,45)      1

=== 会话窗口（gap=5s，按用户，按事件时间）===
user A: 3 个会话 [[2 4] [11 21] [33 33]]
user B: 4 个会话 [[3 3] [12 12] [22 22] [35 35]]

结论：水位延迟 0 时 [0,10) 只有 2 条（B3 与 A9 被丢）；延迟 5s 补回 B3，但 A9 仍丢；
      延迟 20s 全部收齐，代价是 [0,10) 的结果要等到 34s 才最终确定。

读数：

延迟	`[0,10)` 计数	丢弃	最大产出延迟	为什么
0s	2	B3、A9	5s	A14 到达时 `W=14`，`[0,10)` 立刻关闭，B3（17s 才到）被丢
5s	3	A9	14s	A14 时 `W=9`，窗口保持打开；A19 到（`W=14`）才关闭，B3 被收进来
20s	4	0	24s	A33 到达（34s）才关窗；代价是 `[0,10)` 等到 34s 才最终确定

🔑 水位线是用延迟换完整度的旋钮。本例把延迟从 0 调到 20s，[0,10) 从 2 条变成 4 条，代价是结果晚 24s。关键是先量出「乱序/迟到分布」，再定延迟，而不是拍一个「5 秒」。

滑动窗口：重叠带来的状态放大

窗口起点	计数	包含
[0,10)	3	A2,A4,B3
[5,15)	3	A11,A14,B12
[10,20)	4	A11,A14,B12,A19
[15,25)	3	A19,A21,B22
[20,30)	2	A21,B22
[25,35)	1	A33
[30,40)	2	A33,B35
[35,45)	1	B35

一条事件会落在 size/slide 个窗口里（本例 10/5 = 2）。如果 size=1h, slide=1s，每条事件要更新 3600 个窗口——状态和计算量都会爆炸。生产上滑动窗口的步长通常不小于分钟级，且要配状态 TTL。

窗口状态属于 Keyed State，状态后端、checkpoint 与 TTL 见 62.3 状态管理与容错。

会话窗口：合并才是难点

用户	事件时间	会话
A	2,4,11,14,19,21,33	`[2,4]`、`[11,21]`、`[33,33]`
B	3,12,22,35	`[3,3]`、`[12,12]`、`[22,22]`、`[35,35]`

会话窗口的状态要支持合并：新事件到来时，可能把左右两个会话连起来（例如 gap=5，事件 16 会把 [11,14] 与 [19,21] 合并）。这比滚动窗口的「一个 key 一个计数器」复杂得多，状态后端要支持按 key 的 list/map 合并。

触发、迟到与清理

机制	作用	坑
on-time 触发	水位越过窗口结束时发结果	迟到数据还没到，结果可能被修正
early 触发	水位未到先发「早期结果」	下游要能处理同一窗口多次更新
late 触发	allowed lateness 内迟到数据再触发	状态不能立刻清，保留时间 = lateness
side output	超过 lateness 的数据旁路输出	不配置就静默丢，无法对账
状态 TTL	过期清理窗口状态	TTL 太短，迟到数据无法更新；太长，状态膨胀

⚠️ 窗口关闭后立刻删状态，是最常见的「迟到数据修不了」原因。allowed lateness 的意义不是「多等一会」，而是状态多保留一会；保留期结束前，窗口必须还能被更新。

生产落地：指名 → 可抄 → 可验 → 可退

分区内有序、跨分区乱序的来源见 43.3 顺序与分区。

指名：Flink SQL / DataStream（窗口与水位的主场）；Kafka Streams 的 TimeWindows / SessionWindows（Kafka 生态内轻量方案）；ksqlDB（SQL 化）。

可抄（Flink SQL 1.17+，未验证：本机未安装 Flink）：

-- 滚动 10s
SELECT window_start, window_end, COUNT(*) AS cnt
FROM TABLE(TUMBLE(TABLE events, DESCRIPTOR(ts), INTERVAL '10' SECOND))
GROUP BY window_start, window_end;

-- 滑动 10s / 步长 5s
SELECT window_start, window_end, COUNT(*) AS cnt
FROM TABLE(HOP(TABLE events, DESCRIPTOR(ts), INTERVAL '5' SECOND, INTERVAL '10' SECOND))
GROUP BY window_start, window_end;

-- 会话 gap 5s
SELECT window_start, window_end, user_id, COUNT(*) AS cnt
FROM TABLE(SESSION(TABLE events, DESCRIPTOR(ts), INTERVAL '5' SECOND))
GROUP BY window_start, window_end, user_id;

Kafka Streams 侧：

TimeWindows.ofSizeAndGrace(Duration.ofSeconds(10), Duration.ofSeconds(5)); // 10s 窗口 + 5s 允许迟到
SessionWindows.ofInactivityGapWithNoGrace(Duration.ofSeconds(5));          // 5s 会话间隔

可验：本节程序输出 delay 0/5/20 三行，直接看「丢弃数 / 最大产出延迟」；Flink 侧看 Web UI 的 watermark 与 numLateRecordsDropped 指标，迟到率突然上升通常意味着上游乱序变大或水位延迟设小了。可退：调大 WATERMARK ... INTERVAL / allowed lateness，或把迟到数据侧输出到补偿链路；窗口大小与步长变更要重跑历史（60.2 数仓分层 (ODS / DWD / DWS / ADS) 的可重算原则同样适用）。

反模式

反模式	后果
水位延迟拍脑袋设 1s	乱序事件大量丢弃，指标系统性偏低
水位延迟设成小时级	实时链路退化成批处理，告警永远慢半拍
多分区水位取 max	快分区推过慢分区，慢分区数据被丢
滑动窗口 slide 设成 1s	每条事件更新上千个窗口，状态与 CPU 爆炸
窗口关闭立即清状态	迟到数据无法更新，历史窗口永远错
会话窗口不支持合并	同一个用户的会话被切成多段
窗口结果只 append	同一窗口多条记录，查询重复计数
只看窗口大小不看业务周期	1 分钟窗口跨不过结算周期，指标无意义

自检

Q1 滚动、滑动、会话窗口分别回答什么业务问题？各举一个例子。

▸ 答案：滚动窗口回答「固定周期内发生了什么」（每分钟 GMV、每小时订单量）；滑动窗口回答「最近一段时间怎么样」（最近 10 分钟失败率、5 分钟移动平均）；会话窗口回答「一次行为从哪里开始、到哪里结束」（用户点击流、设备在线段）。选窗口的依据是业务问题的语义，不是「哪个实现简单」。

Q2 本节把水位延迟从 0 调到 5s，[0,10) 的计数为什么从 2 变成 3？

▸ 答案：延迟 0 时，A14（到达 15s）把最大事件时间推到 14，水位 W=14，[0,10) 立即关闭；B3 在 17s 才到，已被判迟到丢弃。延迟 5s 时，A14 到达后 W=9，[0,10) 仍开着；直到 A19 到达（W=14）才关闭，此时 B3 已经进来，所以计数从 2 变 3。

Q3 延迟 20s 时为什么 [0,10) 能收到 4 条，代价是什么？

▸ 答案：延迟 20s 让水位推进很慢，[0,10) 直到 A33 到达（34s，水位 13s）才关闭，A9（24s 到）也被收进来，所以是 A2、A4、B3、A9 四条。代价是结果产出延迟 24s（窗口结束 10s，最终确定 34s），实时性下降。

Q4 多输入算子的水位为什么要取 min？空闲分区会怎样？

▸ 答案：输出水位代表「所有输入都至少到这里」，取 max 会让快输入把水位推过慢输入，慢输入中事件时间较早的数据会被误判迟到。取 min 保证没有输入被落下。空闲分区没有新事件，水位不推进，会把全局水位拖住；要用 idle source 标记（Flink 的 withIdleness / table.exec.source.idle-timeout）把它排除。

Q5 allowed lateness 与 side output 的区别是什么？

▸ 答案：allowed lateness 是「窗口关闭后仍保留状态一段时间」，迟到数据可以继续更新窗口并重新触发结果；side output 是「超过 lateness 的数据不再进入主窗口，而是写到旁路流」。前者修正主结果，后者保留证据/离线补偿。生产上通常两者都要：lateness 覆盖常见迟到，side output 兜底极端迟到。

Q6 滑动窗口 size=1h, slide=1s 为什么会把状态打爆？

▸ 答案：每条事件属于 size/slide = 3600 个窗口，每个窗口都要维护计数/聚合状态；一个 key 每秒一条事件，就要更新 3600 个窗口，状态条目数和 CPU 都按 3600 倍放大。滑动窗口的步长要按业务精度定，通常分钟级；更细的滑动需求用预聚合/降采样，而不是直接开小 slide。

Q7 窗口大小和会话 gap 应该怎么定？

▸ 答案：窗口大小由业务周期和决策频率决定（日报用天、实时监控用分钟），且要能整除业务周期，避免跨周期边界；会话 gap 由「用户两次行为间隔」的分布决定，通常取 P90–P95 的间隔，太小会把一次会话切碎，太大则把无关行为合并。定完要用历史数据回放验证，而不是拍一个整数。