⏱️ 03 · Flink 与实时计算：窗口 / Watermark / 状态
> 实时计算岗位必考 **Flink**。本节手写核心概念：**事件时间、水位线（Watermark）、
> 滚动/滑动窗口、迟到数据处理**，用 numpy 模拟流式处理（无 Flink 依赖）。
📋 本节要解决的问题
| # | 问题 | 方法 |
|---|---|---|
| 1 | 批与流区别？ | 有界 vs 无界数据 |
| 2 | 事件时间 vs 处理时间？ | 乱序怎么处理 |
| 3 | 水位线（watermark）？ | 允许延迟的边界 |
| 4 | 窗口类型？ | 滚动/滑动/会话 |
%matplotlib inline
import matplotlib
import matplotlib.pyplot as plt
matplotlib.rcParams['font.sans-serif'] = ['Microsoft YaHei', 'SimHei', 'PingFang SC', 'Noto Sans CJK SC', 'Arial Unicode MS']
matplotlib.rcParams['axes.unicode_minus'] = False
import numpy as np
import pandas as pd
import sqlite3
import os
rng = np.random.default_rng(42)
1️⃣ 批处理 vs 流处理（必背）
print('''批处理（离线）：数据有界、一次性处理、延迟高（小时级）、准确；
流处理（实时）：数据无界、持续到达、延迟低（秒级）、对乱序敏感；
Lambda 架构：批+流两套并行再合并（历史准确+实时快但对账难）；
Kappa 架构：只用流处理（重放历史=批），Flink 主流。
''')批处理（离线）：数据有界、一次性处理、延迟高（小时级）、准确；
流处理（实时）：数据无界、持续到达、延迟低（秒级）、对乱序敏感；
Lambda 架构：批+流两套并行再合并（历史准确+实时快但对账难）；
Kappa 架构：只用流处理（重放历史=批），Flink 主流。
2️⃣ 三种时间与乱序问题
三种时间语义（必背）
| 时间 | 定义 | 用途 |
|---|---|---|
| 事件时间 Event Time | 事件**发生**的时刻（数据里带的时间戳） | 业务正确性首选 |
| 摄入时间 Ingestion Time | 进入 Flink 的时刻 | 少用 |
| 处理时间 Processing Time | 算子**处理**它的时刻（看机器时钟） | 低延迟、可容忍近似 |
**乱序是常态**：网络抖动、分区并行都会让"先发生的数据后到达"。
若按处理时间统计"10:00:00–10:00:10 的订单"，一条 10:00:09 发出、10:00:11 才到的订单会被算错窗口。
所以严肃的实时统计必须用**事件时间 + Watermark**。
# 事件时间（事件发生时刻）vs 处理时间（数据到达处理系统时刻）
events = [
('A', '2024-01-01 10:00:01'),
('B', '2024-01-01 10:00:02'),
('D', '2024-01-01 10:00:04'),
('C', '2024-01-01 10:00:03'), # 乱序：C 在 D 之后才到
]
print('到达顺序:', [e[0] for e in events])
print('事件时间顺序:', sorted(events, key=lambda e: e[1]))
print('''
如果按"处理时间"聚合，会得到错误顺序（C 与 D 顺序反了）；
所以流计算默认用"事件时间 + 水位线"处理乱序。
''')到达顺序: ['A', 'B', 'D', 'C']
事件时间顺序: [('A', '2024-01-01 10:00:01'), ('B', '2024-01-01 10:00:02'), ('C', '2024-01-01 10:00:03'), ('D', '2024-01-01 10:00:04')]
如果按"处理时间"聚合，会得到错误顺序（C 与 D 顺序反了）；
所以流计算默认用"事件时间 + 水位线"处理乱序。
3️⃣ 滚动窗口（Tumbling Window）手写
# 滚动窗口：固定长度、不重叠、每条数据只属于一个窗口
def tumbling_window(events_with_ts, win_len):
# events_with_ts: [(id, ts_seconds)]
buckets = {}
for eid, ts in events_with_ts:
w = (ts // win_len) * win_len # 窗口起点
buckets.setdefault(w, []).append((eid, ts))
return [sorted(v) for k, v in sorted(buckets.items())]
stream = [('a', 3), ('b', 5), ('c', 8), ('d', 12), ('e', 17)]
wins = tumbling_window(stream, 10)
for w in wins:
print(f'窗口 [{w[0][1]}s, {w[0][1]+10}s): {[e[0] for e in w]}')
assert sum(len(w) for w in wins) == len(stream), '滚动窗口每条数据只属于一个窗口'
print('✅ 滚动窗口：每条数据属于唯一窗口')窗口 [3s, 13s): ['a', 'b', 'c']
窗口 [12s, 22s): ['d', 'e']
✅ 滚动窗口：每条数据属于唯一窗口
4️⃣ 滑动窗口（Sliding Window）手写
# 滑动窗口：长度+滑动步长，数据可属于多个窗口
def sliding_window(events_with_ts, win_len, slide):
buckets = {}
for eid, ts in events_with_ts:
start = 0
while start <= ts: # 窗口起点不能晚于事件时间
if start <= ts < start + win_len: # 事件落在窗口内
buckets.setdefault(start, []).append((eid, ts))
start += slide
return buckets
stream2 = [('a', 2), ('b', 5), ('c', 7), ('d', 11)]
buckets = sliding_window(stream2, 6, 3)
for start in sorted(buckets):
print(f'窗口 [{start},{start+6}): {[e[0] for e in buckets[start]]}')
assert sum(len(v) for v in buckets.values()) >= len(stream2), '滑动窗口数据可属于多个窗口'
print('✅ 滑动窗口：同一数据属于多个窗口（重叠）')窗口 [0,6): ['a', 'b']
窗口 [3,9): ['b', 'c']
窗口 [6,12): ['c', 'd']
窗口 [9,15): ['d']
✅ 滑动窗口：同一数据属于多个窗口（重叠）
会话窗口（Session Window）补充
会话窗口没有固定长度：以**活动间隔 gap** 切分——每来一条数据就更新"最近活动时间"，
若相邻数据间隔超过 gap（如 15 分钟未操作），则认为会话结束，窗口关闭。
典型用途：用户行为会话分析（一次会话内的浏览/点击合并统计）。
对比记忆：**滚动=切片不重叠；滑动=固定步长重叠；会话=按活动间隙动态切**。
# 可视化：滚动 vs 滑动窗口（时间轴）
fig, axes = plt.subplots(2, 1, figsize=(10, 5), sharex=True)
ts = np.arange(0, 26)
for ax, kind, fc, ec in [(axes[0], '滚动窗口: 长10, 步长=10（不重叠）', '#FBE5D6', '#ED7D31'),
(axes[1], '滑动窗口: 长10, 步长5（重叠，数据可属于2个窗口）', '#E2EFDA', '#548235')]:
ax.set_ylim(0, 1.7); ax.set_yticks([])
ax.scatter(ts, [0.3] * len(ts), s=22, color='#2F5597', zorder=3)
step = 10 if kind.startswith('滚动') else 5
for s in range(0, 26, step):
ax.add_patch(plt.Rectangle((s, 0.7), 10, 0.6, fc=fc, ec=ec, alpha=0.8))
ax.text(s + 5, 1.0, f'[{s},{s+10})', ha='center', fontsize=8.5)
ax.set_title(kind, fontsize=11, loc='left')
ax.set_xlim(-1, 27)
axes[1].set_xlabel('事件时间（秒）')
fig.suptitle('蓝点=事件；滚动窗口每条事件只属 1 个窗口，滑动窗口可属多个', fontsize=12)
plt.show()
5️⃣ Watermark：允许乱序的边界
# Watermark = 当前"已安全"的事件时间下限
# 含义：watermark = max(已见事件时间) - 允许乱序延迟
def watermark_update(seen_max_ts, max_lateness):
return seen_max_ts - max_lateness
events_ts = [10, 12, 11, 15, 14, 20, 19] # 乱序到达
allow_late = 3
wm = 0
print('到达时间 | 当前 watermark')
for ts in events_ts:
wm = max(wm, watermark_update(ts, allow_late))
print(f' {ts:>4} | {wm:>4}（=最新事件时间-{allow_late}，窗口关闭判断依据）')
print('''
判断：事件时间 >= watermark 的事件才允许进入窗口计算；
小于 watermark 的事件视为迟到 -> 侧输出/丢弃/重算。
''')到达时间 | 当前 watermark
10 | 7（=最新事件时间-3，窗口关闭判断依据）
12 | 9（=最新事件时间-3，窗口关闭判断依据）
11 | 9（=最新事件时间-3，窗口关闭判断依据）
15 | 12（=最新事件时间-3，窗口关闭判断依据）
14 | 12（=最新事件时间-3，窗口关闭判断依据）
20 | 17（=最新事件时间-3，窗口关闭判断依据）
19 | 17（=最新事件时间-3，窗口关闭判断依据）
判断：事件时间 >= watermark 的事件才允许进入窗口计算；
小于 watermark 的事件视为迟到 -> 侧输出/丢弃/重算。
# 可视化：Watermark 单调推进与迟到数据（阶梯线）
arrive = [1, 4, 2, 7, 3] # 按到达顺序的事件时间
allow = 2
wm, hist, late = 0, [], []
for t in arrive:
new_wm = max(wm, t - allow)
hist.append(new_wm)
late.append(t < new_wm)
wm = new_wm
fig, ax = plt.subplots(figsize=(9.5, 4))
ax.step(range(len(arrive)), hist, where='post', color='#2F5597', lw=2,
label='Watermark = max(已见事件时间) - 2')
ax.plot(range(len(arrive)), arrive, 'o', color='#548235', ms=8, label='事件时间(到达顺序)')
for i, t in enumerate(arrive):
if late[i]:
ax.plot(i, t, 'X', color='#C00000', ms=13, mec='k', zorder=5)
ax.annotate('迟到->侧输出/丢弃', (i, t), textcoords='offset points', xytext=(-60, 16),
fontsize=10, color='#C00000', arrowprops=dict(arrowstyle='->', color='#C00000'))
ax.annotate(f'{t}s', (i, t), textcoords='offset points', xytext=(0, 8), ha='center', fontsize=9, color='#548235')
ax.set_xticks(range(len(arrive))); ax.set_xticklabels([f'第{i+1}条' for i in range(len(arrive))])
ax.set_ylim(-0.5, 8.5); ax.legend(fontsize=9.5); ax.grid(axis='y', alpha=0.3)
ax.set_title('Watermark 推进：第5条 t=3s 已低于水位线 5s -> 迟到', fontsize=12)
plt.show()
6️⃣ 迟到数据处理策略（面试速答）
状态、Checkpoint 与精确一次（面试深挖版）
**状态（State）**：算子在处理流数据时要"记住"的中间结果（如窗口内累加值、去重集合）。
分**算子状态**与**键控状态（Keyed State）**，存储在**状态后端**（堆内存 HashMap / RocksDB，
后者支持超大状态）。
**Checkpoint（一致性快照）**：基于 **Chandy-Lamport 异步屏障快照**——
Checkpoint 协调器向源头注入**屏障（barrier）**，屏障随数据流向下游；
算子收到全部输入的屏障后把当前状态**快照**到远程存储，然后继续处理数据（**几乎不阻塞计算**）。
故障恢复：所有算子从最近一次成功快照恢复状态 + 重放源数据。
**精确一次（Exactly-once）**＝ Checkpoint 恢复 + **Sink 端两阶段提交**：
屏障到来时 Sink 把数据**预提交**（写入但不对外可见），Checkpoint 完成后**正式提交**。
配合 Kafka 事务，可做到"故障也不多算、不少算"。
print('''1) 允许迟到（allowedLateness）：窗口关闭后再等一段时间；
2) 侧输出流（sideOutput）：迟到数据单独输出，离线补偿；
3) 直接丢弃：最简粗暴，准确性差；
4) 水印合并（多并行度）：取最小 watermark（最保守）；
5) 面试要点：watermark 决定"窗口何时关"，allowedLateness 决定"关了还能不能补"。
''')1) 允许迟到（allowedLateness）：窗口关闭后再等一段时间；
2) 侧输出流（sideOutput）：迟到数据单独输出，离线补偿；
3) 直接丢弃：最简粗暴，准确性差；
4) 水印合并（多并行度）：取最小 watermark（最保守）；
5) 面试要点：watermark 决定"窗口何时关"，allowedLateness 决定"关了还能不能补"。
7️⃣ Flink 与 Spark Streaming 对比
print('''| 维度 | Flink | Spark Streaming |
|---|---|---|
| 模型 | 真正的流（每条处理） | 微批次（秒级攒批） |
| 延迟 | 毫秒-秒 | 秒级（批延迟） |
| 状态 | 原生状态管理（checkpoint） | 有状态但较粗 |
| 精确一次 | Exactly-once（两阶段提交） | 有（2.0+ 结构化流） |
| 适用 | 实时风控/实时大屏 | 近实时批处理 |
结论：新增实时项目普遍选 Flink；Spark 流适合"批+流一体"需求。
''')| 维度 | Flink | Spark Streaming |
|---|---|---|
| 模型 | 真正的流（每条处理） | 微批次（秒级攒批） |
| 延迟 | 毫秒-秒 | 秒级（批延迟） |
| 状态 | 原生状态管理（checkpoint） | 有状态但较粗 |
| 精确一次 | Exactly-once（两阶段提交） | 有（2.0+ 结构化流） |
| 适用 | 实时风控/实时大屏 | 近实时批处理 |
结论：新增实时项目普遍选 Flink；Spark 流适合"批+流一体"需求。
📝 自测清单
• [ ] 能对比批/流、Lambda/Kappa
• [ ] 能解释事件时间 vs 处理时间与乱序
• [ ] 能手写滚动/滑动窗口
• [ ] 能解释 watermark 的作用与计算
• [ ] 能说出迟到数据三种处理策略