如何设计动态 Top-K 高频元素数据结构？

题干与适用场景

一个整数 ID 流持续到达。add(x) 记录 ID x 新出现一次；topK(k) 返回当前频次最高的至多 k 个不同 ID 及其频次，同频 ID 的顺序不限，k 每次查询可以不同。请为读多写少和写多读少两种负载设计精确方案，分析时间与空间复杂度；若不同 ID 数量可能无限增长且内存固定，再给出近似方案及其误差含义。

本题面向软件工程和算法岗位。精确部分只处理频次递增，不包含删除、滑动窗口或分布式合并。令 N 为累计更新数，D 为不同 ID 数，且通常 k << D。公开面试题原文同时要求比较读写负载，并把有限内存流式近似作为扩展，因此不能只背“哈希表加最小堆”。

面试官考察点

第一，候选人是否先确认操作契约和负载比例。一个只在结束时查询一次的任务，与每次写入后都查询的在线排行榜，不应使用同一维护成本。

第二，能否给出复杂度与状态一致的方案。哈希表计数后在查询时建大小为 k 的最小堆，add 的期望时间为 O(1)，topK 为 O(D log k)；若声称每次更新堆都是 O(log k)，还必须维护堆内位置，并解释堆外元素何时进入。

第三，能否证明动态结构正确。强回答会明确三个不变量：每个 ID 只属于一个频次桶；桶频次严格递增且没有空桶；ID 所在桶的频次等于它的真实累计次数。复杂度结论必须依赖这些不变量。

第四，能否区分精确 top-k、重频项和频次估计。Space-Saving 可以在固定计数器下给出候选项及上下界；Count-Min Sketch 主要回答给定 ID 的近似频次，本身不保存可枚举的 ID 集合。把 sketch 单独当作 top-k 列表会漏掉“从哪里取得候选键”这一步。

回答前需要澄清的问题

• k 固定还是每次可变？ 固定 K 可以维护带位置索引的大小 K 堆；任意 k 更适合按频次有序的结构。
• 更新与查询比例是多少？ 写多查少时把工作延后到查询更省维护成本；查询频繁时值得在每次 add 中维护顺序。
• 同频项需要确定顺序吗？ 本题允许任意顺序，所以桶内用哈希集合即可；若要求 ID 升序，桶内要换成有序集合，更新成本不再是期望 O(1)。
• 是否有删除或时间窗口？ 只有递增时，元素只从频次 f 移到相邻的 f + 1；删除会增加反向移动，窗口还需要保存过期事件。
• D 是否能放入内存？ 精确答案在一般分布下需要保留所有不同 ID 的频次。内存固定时必须接受近似结果或增加可重放的第二遍。
• 允许何种误差？ “大致正确”无法验收。应明确是频次加性误差、候选集合可能不确定，还是在频次间隔足够大时保证 top-k。
• 计数会不会溢出？ 长期服务需要 64 位或更宽计数器；示例使用 JavaScript number，只在安全整数范围内保证精确计数。

30 秒回答框架

“我会先确认 k 是否可变、读写比例、同频排序和内存上限。写多查少时，用哈希表做期望 O(1) 的 add，查询时扫描 D 个频次并维护大小 k 的最小堆，成本 O(D log k)。若任意 k 的查询很频繁，我会维护按频次递增的双向桶链表和 ID → 桶 哈希表；一次更新只把 ID 从 f 桶移到相邻的 f + 1 桶，所以期望 O(1)，从尾部取 k 个结果为 O(min(k, D))，空间 O(D)。若 D 无法放入内存，我会用 Space-Saving 保留固定数量的带误差计数器，只在上下界分离时声称 top-k 有保证；Count-Min Sketch 还需要额外候选集合才能枚举答案。”

分步骤深入解答

第一步：先列出随负载改变的精确方案

“哈希表 + 最小堆”不是唯一答案。它把排序成本放在查询端，恰好适合写多查少。若每次 add 后都要展示排行榜，反复扫描 D 个键会成为瓶颈，频次桶才值得承担更高的实现复杂度。

第二步：建立频次桶不变量

桶链表按频次从小到大排列；每个桶保存该频次下的一组 ID。另一个哈希表把 ID 直接定位到桶。新增 ID 进入频次 1 桶；已有 ID 从 f 桶移动到 f + 1 桶。因为一次更新只增加 1，新桶只可能位于原桶与其后继之间，不需要搜索整条链表。旧桶变空时立即删除。

三个不变量足以证明结果：

1. locations 中每个 ID 恰好出现在一个桶的集合里。
2. 每个非空桶的 frequency 等于其中所有 ID 的真实频次。
3. 从 head 到 tail 的频次严格递增。

因此从 tail 向前枚举时，任何尚未输出的 ID 都不会比已输出 ID 更高频；同频顺序由题目允许任意。每个经过的桶至少输出一个元素，所以查询访问的桶数不超过输出数，时间为 O(min(k, D))。

第三步：实现精确的任意 k 查询

interface Bucket {
  frequency: number;
  values: Set<number>;
  prev: Bucket | null;
  next: Bucket | null;
}

interface TopKEntry {
  value: number;
  count: number;
}

class FrequencyIndex {
  private readonly locations = new Map<number, Bucket>();
  private head: Bucket | null = null;
  private tail: Bucket | null = null;

  add(value: number): void {
    const source = this.locations.get(value);

    if (!source) {
      let target = this.head;
      if (!target || target.frequency !== 1) {
        target = this.insertBefore(this.head, 1);
      }
      target.values.add(value);
      this.locations.set(value, target);
      return;
    }

    let target = source.next;
    if (!target || target.frequency !== source.frequency + 1) {
      target = this.insertAfter(source, source.frequency + 1);
    }

    source.values.delete(value);
    target.values.add(value);
    this.locations.set(value, target);

    if (source.values.size === 0) {
      this.removeBucket(source);
    }
  }

  topK(k: number): TopKEntry[] {
    if (!Number.isInteger(k) || k < 0) {
      throw new RangeError("k must be a non-negative integer");
    }

    const result: TopKEntry[] = [];
    let bucket = this.tail;

    while (bucket && result.length < k) {
      for (const value of bucket.values) {
        result.push({ value, count: bucket.frequency });
        if (result.length === k) break;
      }
      bucket = bucket.prev;
    }

    return result;
  }

  private insertBefore(next: Bucket | null, frequency: number): Bucket {
    const bucket: Bucket = {
      frequency,
      values: new Set<number>(),
      prev: next?.prev ?? null,
      next,
    };

    if (bucket.prev) bucket.prev.next = bucket;
    else this.head = bucket;

    if (next) next.prev = bucket;
    else this.tail = bucket;

    return bucket;
  }

  private insertAfter(prev: Bucket, frequency: number): Bucket {
    const bucket: Bucket = {
      frequency,
      values: new Set<number>(),
      prev,
      next: prev.next,
    };

    if (prev.next) prev.next.prev = bucket;
    else this.tail = bucket;

    prev.next = bucket;
    return bucket;
  }

  private removeBucket(bucket: Bucket): void {
    if (bucket.prev) bucket.prev.next = bucket.next;
    else this.head = bucket.next;

    if (bucket.next) bucket.next.prev = bucket.prev;
    else this.tail = bucket.prev;
  }
}

代码的复杂度使用 Map 与 Set 的常见期望常数时间假设，并非 JavaScript 规范承诺的严格最坏 O(1)。topK(0) 返回空数组，k > D 返回全部 ID，非法的负数或非整数 k 抛出错误。

第四步：内存固定时明确近似保证

精确哈希表的空间随 D 增长。Space-Saving 改为只维护 m 个“ID、估计频次、最大误差”计数器，并要求 m > k 才能比较第 k + 1 个边界候选：已监控 ID 到达时递增；未监控 ID 到达且计数器已满时，替换当前最小估计频次为 cmin 的 ID，把新 ID 的估计值设为 cmin + 1，误差记为 c_min。

对每个仍被监控的 ID，真实频次位于 [估计值 - 误差, 估计值]；论文给出的最大高估不超过 N / m。若前 k 个候选的最小下界都不小于第 k + 1 个候选的估计上界，就能根据当前摘要证明 top-k 集合；边界交叠时只能报告近似候选，不能把估计排序伪装成精确答案。若整个流的 D <= m，没有发生替换，计数就是精确的。

Count-Min Sketch 使用固定的 width × depth 计数数组。取 width = ceil(e / ε)、depth = ceil(ln(1 / δ)) 时，对只增不减的流，给定 ID 的估计值不会低于真实值，并以至少 1 - δ 的概率不超过 真实值 + εN。它不保存 ID，因此还要结合候选堆、候选集合或可枚举域；只建 sketch 无法直接回答“有哪些 top-k”。

第五步：用对照实现验证，而不只测示例

先用序列 [1, 2, 1, 3, 2, 1] 验证 topK(2) 返回频次 3 和 2 的两个 ID。再覆盖空结构、k = 0、k > D、全部同频、单一热点加大量一次性 ID，以及一个 ID 连续跨越多个桶。

最后生成随机更新流，以朴素哈希计数加完整排序作为 oracle。每隔若干次更新比较：返回长度为 min(k, D)；ID 不重复；每个频次准确；任何未选 ID 的频次不高于已选集合的最小频次。本文实现已用 10,000 次确定性随机更新和多个 k 值完成这一对照。

高质量示范回答

“我先把契约限定为只增不减、k 每次可变、同频任意顺序。若写入很多而查询很少，我只维护 ID → 频次 哈希表，add 期望 O(1)；查询时扫描 D 个 ID，用大小 k 的最小堆，时间 O(D log k)、额外空间 O(k)。

如果排行榜查询频繁，我会用双向频次桶。桶从低频到高频排列，每个桶放同频 ID，再用哈希表把 ID 定位到桶。一次 add 只会从 f 移到 f + 1，因此只检查相邻桶，空桶立即删除。这样更新是期望 O(1)，从尾桶向前取结果为 O(min(k, D))，总空间 O(D)。正确性依赖每个 ID 唯一归桶、桶值等于真实频次、桶序严格递增。

若 D 无法放进内存，精确契约必须改变。我会用 m 个 Space-Saving 计数器保存候选和误差区间，最大高估由 N / m 控制；只有前 k 个下界与后续上界分离时才宣称集合有保证。Count-Min Sketch 可以估计给定 ID 的频次，但还需要候选发现结构。上线前我会用朴素完整排序做随机差分测试，并单测同频、非法 k 和计数器溢出边界。”

常见错误

• 不问负载就固定使用最小堆 → 查询频繁时每次扫描全部 D 个 ID，写多查少时又可能过度维护 → 先按更新与查询比例决定把成本放在哪一端。
• 把可变 k 当固定 K 维护 → 大于预维护 K 的查询没有完整候选 → 明确 k 上限，或使用能按任意 k 遍历的频次桶或有序结构。
• 堆中直接修改频次 → 普通堆不知道元素位置，旧键会破坏堆序或产生大量陈旧记录 → 维护 ID → 堆位置，或删除旧键后重新插入，并写清复杂度。
• 频次桶变空后不删除 → 查询可能遍历从 1 到最大频次的大量空洞 → 每次移动后立即摘除空桶，保持只链接非空频次。
• 宣称哈希操作严格 O(1) → Map 与 Set 只支持常见期望复杂度分析 → 说明哈希假设；需要最坏界限时选择平衡树并接受 O(log D)。
• 用 Count-Min Sketch 后直接返回 top-k → sketch 只接受给定键的查询，不能枚举未知 ID → 同时维护候选发现结构，或改用保存候选键的 Space-Saving。
• 报告近似结果却不附误差 → 面试官无法判断第 k 与第 k + 1 是否可区分 → 返回估计值、误差上下界和能否保证集合的结论。
• 只测固定示例 → 链表断链、空桶和同频边界常在长更新序列中出现 → 用朴素排序 oracle 做随机差分测试并检查不变量。

追问及应对

追问一：如果 topK 永远只查询 K = 100，还需要频次桶吗？

不一定。维护哈希频次、大小 100 的最小堆和 ID → 堆位置，每次更新后调整堆内元素或比较堆顶，成本 O(log 100)，实现和内存布局可能更简单。它不能回答 topK(1000)；频次桶的价值在于任意 k 和期望常数更新。

追问二：同频时必须按 ID 升序怎么办？

把桶内 Set 换成有序集合，或在查询时对被截断的边界桶排序。前者让每次移动增加 O(log s)，其中 s 是桶大小；后者只在查询时支付边界桶排序成本。应根据确定顺序是否每次都需要来选择。

追问三：增加 remove(x) 后如何修改？

ID 的频次从 f 降到 f - 1，对称地检查前驱桶，降为 0 时从 locations 删除。还要定义删除不存在 ID 是报错还是忽略。若同时存在并发 add/remove，需要把“定位、移桶、删空桶”作为一个原子临界区，否则同一 ID 可能出现在两个桶。

追问四：要求最近 10 分钟的 top-k 呢？

频次不再单调。除了桶结构，还要保存带时间的事件或分桶计数，以便过期时执行反向更新。逐事件队列精确但空间按窗口事件数增长；时间分桶降低状态量，却把边界精度变成显式误差。Space-Saving 的全历史摘要不能直接减去任意过期事件。

追问五：Space-Saving 返回的第 k 与第 k + 1 区间重叠怎么办？

增加计数器 m、接受“候选集合未确定”，或在数据可重放时对候选再做一遍精确计数。第二遍只能校准已保留候选；若摘要容量不足以保证真实 top-k 已进入候选集，重放前必须先扩大候选集合。

追问六：数据分布在多个分片，怎样得到全局 top-k？

各分片的局部 top-k 不能在任意分布下直接合并成精确全局 top-k，因为一个在每个分片都略低于局部门槛的 ID，汇总后可能进入全局前列。精确方案需要聚合所有相关频次或维护可证明覆盖的候选上界；近似方案则合并可合并摘要，并把额外误差和延迟写入契约。