迟到数据

工作原理

流处理器根据观察到的最大时间戳推进事件时间水位线。窗口触发后输出结果。之后到达且属于该窗口的事件就是迟到数据。处理器可以丢弃它，在允许延迟期内接受并重新触发窗口，把它路由到旁路输出，或留给批处理修正作业。

迟到数据不同于乱序数据。事件可以在水位线关闭窗口之前乱序到达并被正常计数。问题始于窗口已经产出结果，因为下游系统已经消费了一个现在不完整的结果。

选择策略

先度量：统计迟到事件数量、延迟分布以及它们来自哪些源。如果延迟只有几秒且业务能容忍小幅修正，就允许延迟，并用明确的撤回或 upsert 键输出更新结果。如果延迟可能长达数小时，就使用批处理对账路径，或使用能接受迟到修正的可变服务表。如果迟到事件不可接受，就强制生产者时间戳，并拒绝超出定义范围的事件。

无论采用哪种策略，都要让它可观测。静默丢弃与源根本没有发送事件无法区分。

权衡

允许延迟会增加状态大小和结果抖动，因为窗口保持打开并可能多次输出。丢弃能保持状态小，但损失准确性。重新处理一批数据准确且容易推理，却增加延迟，并且如果输出是追加式的，还会有重复输出风险。

示例与误解

移动应用离线时排队签到，恢复连接后再上传。若管道丢弃一分钟以前的数据，就会丢失合法访问。更好的设计保留 24 小时允许延迟，以窗口和门店为键写入 upsert 结果，并运行夜间批处理来结算当天数据。

误解是迟到数据罕见或恶意。实际上，普通移动网络和生产端重试使迟到成为正常运营状况。

代码示例

-- 每个邮箱保留 created_at 最新的一行，标记其余为重复
WITH ranked AS (
  SELECT
    id,
    email,
    created_at,
    ROW_NUMBER() OVER (
      PARTITION BY email
      ORDER BY created_at DESC, id DESC
    ) AS rn
  FROM subscribers
)
SELECT id, email, created_at, (rn = 1) AS is_primary
FROM ranked
ORDER BY email, rn;

-- 一次性清扫：删除全部非首行
WITH ranked AS (
  SELECT id,
         ROW_NUMBER() OVER (
           PARTITION BY email
           ORDER BY created_at DESC, id DESC
         ) AS rn
  FROM subscribers
)
DELETE FROM subscribers s
USING ranked r
WHERE s.id = r.id
  AND r.rn > 1;

-- 上游堵漏：唯一索引阻止重复再次写入
CREATE UNIQUE INDEX CONCURRENTLY idx_subscribers_email
  ON subscribers (email);

export interface CountMinOptions {
  width?: number;
  depth?: number;
  seeds?: readonly number[];
}

export class CountMinSketch {
  private readonly width: number;
  private readonly depth: number;
  private readonly table: Uint32Array;
  private readonly seeds: readonly number[];
  private total = 0;

  constructor(options: CountMinOptions = {}) {
    const width = options.width ?? 2048;
    const depth = options.depth ?? 5;
    if (!Number.isInteger(width) || width <= 0 || !Number.isInteger(depth) || depth <= 0) {
      throw new RangeError("width and depth must be positive integers");
    }
    this.width = width;
    this.depth = depth;
    this.table = new Uint32Array(width * depth);
    this.seeds = options.seeds ?? Array.from(
      { length: depth },
      (_, index) => Math.imul(index + 1, 0x9e3779b1) >>> 0,
    );
    if (this.seeds.length !== depth) {
      throw new RangeError("seeds length must equal depth");
    }
  }

  add(value: string, count = 1): this {
    if (!Number.isInteger(count) || count < 1) {
      throw new RangeError("count must be a positive integer");
    }
    for (let row = 0; row < this.depth; row += 1) {
      const index = row * this.width + this.hash(value, this.seeds[row]!);
      this.table[index] = (this.table[index]! + count) >>> 0;
    }
    this.total += count;
    return this;
  }

  estimate(value: string): number {
    let minimum = Number.POSITIVE_INFINITY;
    for (let row = 0; row < this.depth; row += 1) {
      minimum = Math.min(
        minimum,
        this.table[row * this.width + this.hash(value, this.seeds[row]!)]!,
      );
    }
    return minimum;
  }

  private hash(value: string, seed: number): number {
    let hash = seed >>> 0;
    for (let index = 0; index < value.length; index += 1) {
      hash ^= value.charCodeAt(index);
      hash = Math.imul(hash, 0x01000193) >>> 0;
    }
    return hash % this.width;
  }

  get count(): number {
    return this.total;
  }
}