题目重点：从精确单机滑动窗到分区流处理，处理用户事件计数、迟到数据、状态过期和 Top 100 汇总。
题意与假设
这道练习题保留了面经中值得训练的核心约束，并将需要澄清的条件单独列出。现场拿到变体时，先和面试官确认这些条件。
- 默认所有 eventType 都计数，时间窗为 (now−3600s,now]；同数按 userId 字典序稳定排序。
- 展示代码是事件按 timestamp 非降序到达的单机精确训练版；生产异步分区会乱序，需事件时间 watermark 和迟到修正。
解题路线
- 单机精确基线用队列保存窗口内 (timestamp,userId)，map 保存每个用户计数；新事件先移除超窗事件再加一。
- 每次查询从活跃用户计数中按次数降序取 Top 100。排序 O(U log U) 简单正确；高查询 QPS 可用可删有序索引或按周期重建榜单，不应把旧堆元素当作仍有效计数。
- 分布式时按 userId 分区计数并持久化状态，窗口过期和 checkpoint 同步；各分区输出局部 Top 100 后可合并，因为全局计数按 userId 完整归属一个分区。
- Flink 或 Spark 的选择取决于延迟、状态量、批流统一和运维能力；先给输入速率、允许延迟、恢复点和准确度，再选框架。
手动走一遍
窗口 10 秒示例：U1@1、U2@2、U1@4，now=4 时 U1:2 领先；now=12 时 U1@1 已过期，U2@2、U1@4 留在窗口，按 userId 规则并列。代码测试使用同一开闭边界。
Python 参考实现
下面的代码按上文假设实现，适合本地运行并追加自己的测试用例。
代码区可横向滚动；使用键盘时，先聚焦代码区再按方向键查看。
"""Exact event-time rolling counts for monotonic input; training reference."""
from collections import deque
class RollingTopK:
def __init__(self, window_seconds: int):
if window_seconds <= 0:
raise ValueError("window_seconds must be positive")
self.window_seconds = window_seconds
self.events: deque[tuple[int, str]] = deque()
self.counts: dict[str, int] = {}
self.now: int | None = None
def _expire(self, now: int) -> None:
cutoff = now - self.window_seconds
while self.events and self.events[0][0] <= cutoff:
_, key = self.events.popleft()
remaining = self.counts[key] - 1
if remaining:
self.counts[key] = remaining
else:
del self.counts[key]
def add(self, timestamp: int, key: str) -> None:
if self.now is not None and timestamp < self.now:
raise ValueError("events must arrive in timestamp order")
self.now = timestamp
self._expire(timestamp)
self.events.append((timestamp, key))
self.counts[key] = self.counts.get(key, 0) + 1
def top_k(self, now: int, k: int) -> list[tuple[str, int]]:
if k < 0:
raise ValueError("k must be nonnegative")
if self.now is not None and now < self.now:
raise ValueError("queries must not move time backwards")
self.now = now
self._expire(now)
return sorted(self.counts.items(), key=lambda pair: (-pair[1], pair[0]))[:k]
复杂度
窗口内 E 条事件、U 个活跃用户：add 均摊 O(1)、单次过期可 O(E) 但每事件只移除一次；top100 排序 O(U log U)、空间 O(E+U)。分布式吞吐与状态规模需按分区倾斜测量。
易错点与追问
容易失分的边界
- 把按固定小时切的 tumbling window 当作任意时刻过去一小时。
- 只做每分区原始事件 Top 100 再合并，若同一用户跨分区则局部计数不完整。
- 忽略迟到事件和去重，导致重放后计数翻倍。
继续深挖
- 如果最热用户成为单分区热点，如何在保持准确性的同时做两级聚合？
- 如果不允许修正历史结果，watermark 等待多久才发布？
