Find Median from Data Stream

Q: 从数据流中查找中位数的时间复杂度是多少？

使用两个堆时， addNum 的时间复杂度为 O(log n) ，最多执行三次堆推入或弹出操作； findMedian 的时间复杂度为 O(1) ，只需读取两次。每个值都会被存储，因此空间复杂度为 O(n) 。而每次查询都排序的话，每次调用 findMedian 的时间复杂度为 O(n log n) 。

Q: 为什么不把这些值保存在已排序的数组中？

排序后的数组可以在 O(1) 时间内回答 findMedian ，二分查找可以在 O(log n) 时间内找到新值应插入的位置。但插入新值会移动该位置之后的所有值，每次 addNum 都需要 O(n) 时间。

困难堆

+10

设计一个包含两个操作的数据结构：addNum 用于存储一个整数，findMedian 用于返回目前存储的所有值的中位数。奇数个值的中位数是中间的值；偶数个值的中位数是中间两个值的平均值。

操作 i 是 ops[i]，其中 values[i] 在操作为 addNum 时保存对应的数字，在操作为 findMedian 时保存 0。在一个空数据结构上按顺序运行这些操作，并为每个操作返回一个字符串：addNum 返回 "null"，findMedian 返回中位数。中位数的表示形式如 "3"、"2.5" 或 "-0.5"：整数不带小数点，零写作 "0"，绝不写作 "-0"。

函数

medianStreamOps(ops: string-array, values: integer-array) → string-array

opsstring-array: 这些操作按运行顺序排列
valuesinteger-array: 每个 addNum 对应的数字，每个 findMedian 对应 0
返回值string-array: 每次操作对应一个答案，格式为文本

约束条件

1 ≤ ops.length ≤ 3000
values.length == ops.length
每个 ops[i] 都是 addNum 或 findMedian。
-10⁵ ≤ values[i] ≤ 10⁵ 对于 addNum，而对于 findMedian，values[i] == 0。
findMedian 只有在至少调用过一次 addNum 后才会被调用。

示例

输入: ops = ["addNum", "findMedian", "addNum", "findMedian", "addNum", "findMedian", "addNum", "findMedian"]values = [5, 0, 1, 0, 8, 0, 3, 0]
输出: ["null", "5", "null", "3", "null", "5", "null", "4"]
解释: 对于 1 和 5，中位数是 (1 + 5) / 2 = 3。对于 1、5 和 8，中位数是 5；对于 1、3、5 和 8，中位数是 (3 + 5) / 2 = 4。

进阶

如果每个值都在 0 到 100 之间，你怎样才能让这两种操作都在常数时间内运行？

提示

逐个打开提示，每个都会多透露一点。

已打开 0/3

题解

中位数位于数据的两半之间，因此每一半内部的顺序并不重要：重要的只有较低一半中的最大值和较高一半中的最小值。两个堆会将这两个值分别置于堆顶。每次调用 addNum 的开销为 O(log n)，每次调用 findMedian 只需以 O(1) 的时间读取两个堆顶。

每次调用 findMedian 时都进行排序

思路

将每个值存入列表。findMedian会对列表排序，并读取索引(total-1)/2和total/2，使用整数除法：元素数量为奇数时，这两个索引相同；为偶数时，它们指向中间的一对元素。将这两个值相加，再把总和除以 2，无论哪种情况都能得到中位数。

使用整数除以 2。偶数总和可以整除。奇数总和则取其绝对值的一半，再加上.5，并保留总和的符号，所以 -3 会变成 -1.5。

每次查询都会重新排序。进行 1,500 次添加，并在每次添加后查询，需要大约 10^7 次比较；而堆只需要几万步。

算法

保留一个列表 values。
对于 addNum x，追加 x 并记录 "null"。
对于 findMedian，对 values 排序，并添加 values[(total-1)/2] 和 values[total/2]。
将总和的一半作为文本处理，使用整数并记录结果。

def half_text(total):
    # total / 2 as text, from integers only: 5 -> "2.5", -1 -> "-0.5", 6 -> "3"
    if total % 2 == 0:
        return str(total // 2)
    sign = "-" if total < 0 else ""
    return sign + str(abs(total) // 2) + ".5"


class MedianFinder:
    def __init__(self):
        self.values = []

    def addNum(self, x):
        self.values.append(x)

    def findMedian(self):
        self.values.sort()
        total = len(self.values)
        # For an odd count both indexes point at the one middle value.
        return half_text(self.values[(total - 1) // 2] + self.values[total // 2])


def medianStreamOps(ops, values):
    finder = MedianFinder()
    result = []
    for op, x in zip(ops, values):
        if op == "addNum":
            finder.addNum(x)
            result.append("null")
        else:
            result.append(finder.findMedian())
    return result

两个堆在中位数处分开

思路

将这些值分成较小的一半 low 和较大的一半 high，确保 low 中的每个值都不大于 high 中的任何值，并且 low 的大小与 high 相同或大 1。中位数是 low 中的最大值；如果值的数量为偶数，则取它与 high 中最小值的平均数。用最大堆存放 low，用最小堆存放 high，这样这些值就会位于堆顶。

如果一个新值不大于 low 的堆顶，就将它放入 low，否则放入 high。当两边的大小不再平衡时，将一边的堆顶移到另一边：它是最接近另一半的值，因此可以维持顺序。一次添加只需要进行几次堆操作，复杂度为 O(log n)；在 R 中则为 O(n)，因为每次更改都会复制堆。

第一个示例添加 4 个值后，low 中有 3 和 1，high 中有 5 和 8。

算法

用最大堆 low 保存较低的一半，用最小堆 high 保存较高的一半。
addNum x：如果 low 为空，或 x 不大于它的堆顶，就把 x 推入 low，否则推入 high。
如果 low 中的值比 high 多两个，就把 low 的堆顶移到 high。如果 high 中的值比 low 多，就把 high 的堆顶移到 low。
findMedian：如果 low 较大，就记录它的堆顶。否则，将两个堆顶相加，再用整数除以 2。

import heapq


def half_text(total):
    # total / 2 as text, from integers only: 5 -> "2.5", -1 -> "-0.5", 6 -> "3"
    if total % 2 == 0:
        return str(total // 2)
    sign = "-" if total < 0 else ""
    return sign + str(abs(total) // 2) + ".5"


class MedianFinder:
    def __init__(self):
        self.low = []   # the smaller half; heapq is a min-heap, so it stores -x to keep the largest on top
        self.high = []  # the larger half, smallest on top

    def addNum(self, x):
        if not self.low or x <= -self.low[0]:
            heapq.heappush(self.low, -x)
        else:
            heapq.heappush(self.high, x)
        # Rebalance: low holds as many values as high, or one more.
        if len(self.low) > len(self.high) + 1:
            heapq.heappush(self.high, -heapq.heappop(self.low))
        elif len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high))

    def findMedian(self):
        if len(self.low) > len(self.high):
            return str(-self.low[0])
        return half_text(-self.low[0] + self.high[0])


def medianStreamOps(ops, values):
    finder = MedianFinder()
    result = []
    for op, x in zip(ops, values):
        if op == "addNum":
            finder.addNum(x)
            result.append("null")
        else:
            result.append(finder.findMedian())
    return result

易错点和边界情况

按大小而不是按值选择堆。直接将值推入较小的堆，而不进行比较，会破坏两部分之间的顺序，堆顶也就不再是中间值。
使用普通整数除法计算一半。在 C 和 Java 中，-1 / 2 的结果是 0，所以追加 .5 后，-0.5 会输出为 0.5。在 Python 中，-3 // 2 的结果是 -2，所以 -1.5 会输出为 -2.5。有些语言使用浮点数时会输出 3.0 或 2.500000。
在最小堆中将 low 取反后存储，却忘记在读取堆顶或将值移到 high 时再取反。
写成 -0。中间值 -1 和 1 的和是 0，答案是 "0"。

常见问题3

从数据流中查找中位数的时间复杂度是多少？

使用两个堆时，addNum 的时间复杂度为 O(log n)，最多执行三次堆推入或弹出操作；findMedian 的时间复杂度为 O(1)，只需读取两次。每个值都会被存储，因此空间复杂度为 O(n)。而每次查询都排序的话，每次调用 findMedian 的时间复杂度为 O(n log n)。

为什么不把这些值保存在已排序的数组中？

排序后的数组可以在 O(1) 时间内回答 findMedian，二分查找可以在 O(log n) 时间内找到新值应插入的位置。但插入新值会移动该位置之后的所有值，每次 addNum 都需要 O(n) 时间。

你能在不存储每个值的情况下，找到数据流的精确中位数吗？

一般来说，不行。到目前为止看到的任何值之后都有可能成为中位数，这取决于接下来会收到什么，因此要得到准确答案，就需要知道所有这些值。

def medianStreamOps(ops, values):
    # 在这里编写代码

测试 #1

测试 #2

参数

ops = ["addNum", "findMedian", "addNum", "findMedian", "addNum", "findMedian", "addNum", "findMedian"]
values = [5, 0, 1, 0, 8, 0, 3, 0]

预期输出

["null", "5", "null", "3", "null", "5", "null", "4"]