Software Engineer / MLE · Coding

TikTok 实时中位数:两堆的秩边界与规模限制

维护较小一半与较大一半,解释 MaxStack 不能直接替代中位数结构。

题目重点:维护较小一半与较大一半,解释 MaxStack 不能直接替代中位数结构。

题意与假设

这道练习题保留了面经中值得训练的核心约束,并将需要澄清的条件单独列出。现场拿到变体时,先和面试官确认这些条件。

  • 训练 append-only 整数流,奇数取中间,偶数取两中间的均值;空流查询错误。
  • 删除、滑窗和近似量化是另一个合同。示例 Fraction 避免偶数均值浮点精度损失。

解题路线

  1. 最大堆 lower 存较小一半,最小堆 upper 存较大一半;所有 lower≤upper,大小差保持0或lower多1。
  2. 按 lower 堆顶决定落哪边,再移动一个堆顶平衡。边界排序和大小不变量一起保证堆顶正是中间秩。
  3. MaxStack 只支持最大端,不维护中间秩,也没有删除最小端的合适操作。扩展它最终仍需增加对半排序结构,不能声称改一个变量就得到 median。
  4. 精确两堆保存所有历史样本,超大流不再满足有限内存;有界值域可频表,允许误差时用 quantile sketch,并声明误差合同。

手动走一遍

流5,1,9,2:中位数依次5、3、5、3.5。负值以取负模拟 maxheap,重复值独立计入排名;极大整数仍可用 Fraction 返回精确均值。

Python 参考实现

下面的代码按上文假设实现,适合本地运行并追加自己的测试用例。

代码区可横向滚动;使用键盘时,先聚焦代码区再按方向键查看。

import heapq
from fractions import Fraction


class RunningMedian:
    """Integer append-only stream; even length returns exact mean of middle pair."""
    def __init__(self):
        self.lower, self.upper = [], []
    def add(self, value):
        if not self.lower or value <= -self.lower[0]:
            heapq.heappush(self.lower, -value)
        else:
            heapq.heappush(self.upper, value)
        if len(self.lower) > len(self.upper) + 1:
            heapq.heappush(self.upper, -heapq.heappop(self.lower))
        elif len(self.upper) > len(self.lower):
            heapq.heappush(self.lower, -heapq.heappop(self.upper))
    def median(self):
        if not self.lower:
            raise IndexError('empty stream')
        if len(self.lower) > len(self.upper):
            return Fraction(-self.lower[0])
        return Fraction(-self.lower[0] + self.upper[0], 2)

复杂度

插入 O(log n),查询 O(1) 结构步骤,空间 O(n)。大整数运算与输出位数另算,分布式合并需不同结构。

易错点与追问

容易失分的边界

  • 只有最大值不足求中位数。
  • 偶数随意取某一项。
  • 声称两堆有限内存处理无限精确流。

继续深挖

  • 滑动窗口删除如何扩展?
  • 近似分位数的误差如何表达?
✓已复制微信号!