水位线

工作原理

事件携带事件时间戳。处理器跟踪已见的最大时间戳，并发出水位线，通常为 max_seen - allowed_delay。算子把水位线与窗口边界比较。当水位线越过窗口结束时间加上允许延迟时，窗口触发，其状态可以被释放。

水位线随记录一起在数据流中传播。多个输入的联合取各输入水位线的最小值，因为最慢的输入仍可能产生旧数据。空闲分区会永远拖住最小值，因此引擎提供空闲检测，将其临时排除。

水位线类型

完美水位线知道未来不可能出现更早的事件，这只有有界输入或协调良好的源才能做到。启发式水位线假设乱序有界，因此可能判断错误。启发式很实用，但意味着窗口触发后仍可能出现迟到事件。有些引擎支持由流中显式标记驱动的标点水位线。

何时使用

当问题关心事情何时发生而不是何时被处理时，使用事件时间窗口和水位线：每小时销售、会话时长、SLA 合规。当新鲜度比事件顺序更重要时，例如检测日志突然激增，可以使用处理时间，并接受结果依赖到达时间。

权衡与示例

允许延迟越大，结果越完整，但延迟更高、保留状态更多；延迟越小，输出越快，但迟到事件更多。水位线还与精确一次接收端相互作用：在迟到数据到达前发出的窗口结果可能需要撤回或更新。

示例：作业按五分钟窗口统计点击，水位线延迟 30 秒。10:05:30 时，10:00 到 10:05 的窗口触发。一个时间戳为 10:04:50、却在 10:05:40 到达的点击属于迟到数据；若允许延迟为五分钟，作业会重新触发该窗口并输出修正后的计数，而不是丢失事件。

代码示例

from collections.abc import Iterable
from dataclasses import dataclass, field


@dataclass
class EmaTracker:
    period: int
    value: float | None = None
    alpha: float = field(init=False, default=0.0)

    def __post_init__(self) -> None:
        if self.period < 1:
            raise ValueError("period must be at least 1")
        self.alpha = 2.0 / (self.period + 1)

    def update(self, sample: float) -> float:
        if self.value is None:
            self.value = sample
        else:
            self.value = self.alpha * sample + (1 - self.alpha) * self.value
        return self.value

    def reset(self) -> None:
        self.value = None


def exponential_moving_average(values: Iterable[float], period: int) -> list[float]:
    tracker = EmaTracker(period)
    return [tracker.update(value) for value in values]


def simple_moving_average(values: list[float], window: int) -> list[float | None]:
    if window < 1:
        raise ValueError("window must be at least 1")
    smoothed: list[float | None] = []
    for index in range(len(values)):
        if index + 1 < window:
            smoothed.append(None)
        else:
            chunk = values[index + 1 - window : index + 1]
            smoothed.append(sum(chunk) / window)
    return smoothed


def main() -> None:
    samples = [10.0, 12.0, 11.0, 15.0, 14.0, 16.0]
    print("ema:", [round(value, 3) for value in exponential_moving_average(samples, 3)])
    print("sma:", simple_moving_average(samples, 3))


if __name__ == "__main__":
    main()

<article>
  <p>Published: <time datetime="2025-01-15">January 15, 2025</time></p>
  <p>Updated: <time datetime="2025-01-16T14:30:00-05:00">Jan 16, 2025 at 2:30 PM</time></p>
  <p>Event: <time datetime="2025-03-01T09:00">March 1, 2025 at 9 AM</time></p>
  <p>Duration: <time datetime="PT2H30M">2 hours 30 minutes</time></p>
  <p>Weekly: Every <time datetime="2025-01-13">Monday</time></p>
</article>

<time datetime="2025-01-15">1/15/2025</time>
<time datetime="2025-01-15T10:00:00Z">10:00 UTC</time>
<time datetime="2025-07-04">Independence Day</time>

<time datetime="PT1H">1 hour</time>
<time datetime="PT30M">30 minutes</time>
<time datetime="P1DT12H">1 day 12 hours</time>

<time datetime="2025-01-15" pubdate>Published Jan 15</time>

<script>
  const times = document.querySelectorAll('time[datetime]');
  times.forEach(t => {
    const date = new Date(t.dateTime);
    const relative = getRelativeTime(date);
    t.title = relative;
  });

  function getRelativeTime(date) {
    const diff = date - new Date();
    const days = Math.round(diff / 86400000);
    return days === 0 ? 'Today' : days > 0 ? `In ${days} days` : `${-days} days ago`;
  }
</script>