考題:資料流的中位數
數字一個一個進來,隨時要答出中位數:每次重新排序、維持排序好的陣列,或用一大一小兩個堆積,成本差了好幾個數量級。
解法
11
題目:數字一個一個進來,每進來一個就要回答「到目前為止的中位數」。例如依序收到 5、15、1、3,中位數是 (3 + 5) / 2 = 4。
資料流
- 68
- 77
- 21
- 62
- 8
- 59
- 72
- 45
- 90
- 22
- 77
1/29
小的那一半:最大堆積(第一格是頂端)
- 68
大的那一半:最小堆積(第一格是頂端)
中位數堆積頂端
68 不大於小半邊的最大值,放進「小的那一半」(最大堆積)。
三種解法跑同一串數字
| 依序回答的中位數 | 總工作量 | 每個數字 | |
|---|---|---|---|
| 每次重新排序 | 68, 72.5, 68, 65, 62, 60.5, 62, 60.5, 62, 60.5, 62 | 313 | O(n log n) |
| 維持排序陣列 | 68, 72.5, 68, 65, 62, 60.5, 62, 60.5, 62, 60.5, 62 | 61 | O(n) |
| 兩個堆積 | 68, 72.5, 68, 65, 62, 60.5, 62, 60.5, 62, 60.5, 62 | 68 | O(log n) |
三種解法在每個數字進來後回答的中位數都相同。
亮起來的是這一步執行的程式碼
function medianResort(stream: number[]): number[] { const seen: number[] = [], out: number[] = []; for (const x of stream) { seen.push(x); const s = [...seen].sort((a, b) => a - b); const m = s.length >> 1; out.push(s.length % 2 ? s[m] : (s[m - 1] + s[m]) / 2); } return out;} function medianSorted(stream: number[]): number[] { const s: number[] = [], out: number[] = []; for (const x of stream) { let lo = 0, hi = s.length; while (lo < hi) { const mid = (lo + hi) >> 1; if (s[mid] <= x) lo = mid + 1; else hi = mid; } s.splice(lo, 0, x); // shifts everything after lo const m = s.length >> 1; out.push(s.length % 2 ? s[m] : (s[m - 1] + s[m]) / 2); } return out;} class Heap { private a: number[] = []; constructor(private less: (x: number, y: number) => boolean) {} get size(): number { return this.a.length; } peek(): number { return this.a[0]; } push(x: number): void { const a = this.a; a.push(x); for (let i = a.length - 1; i > 0; ) { const p = (i - 1) >> 1; if (!this.less(a[i], a[p])) break; [a[i], a[p]] = [a[p], a[i]]; i = p; } } pop(): number { const a = this.a, top = a[0], last = a.pop()!; if (a.length) { a[0] = last; for (let i = 0; ; ) { const l = 2 * i + 1, r = l + 1; let best = i; if (l < a.length && this.less(a[l], a[best])) best = l; if (r < a.length && this.less(a[r], a[best])) best = r; if (best === i) break; [a[i], a[best]] = [a[best], a[i]]; i = best; } } return top; }} function medianHeaps(stream: number[]): number[] { const low = new Heap((a, b) => a > b); // max-heap: smaller half const high = new Heap((a, b) => a < b); // min-heap: larger half const out: number[] = []; for (const x of stream) { if (low.size === 0 || x <= low.peek()) low.push(x); else high.push(x); if (low.size > high.size + 1) high.push(low.pop()); else if (high.size > low.size) low.push(high.pop()); out.push(low.size > high.size ? low.peek() : (low.peek() + high.peek()) / 2); } return out;}模型假設與範圍
- Big O 與實測步數以頁面列出的基本操作為單位;視覺化的快照、畫圖、程式高亮和輸出成本另計,不是執行時間 benchmark。
什麼時候用
- 面試時先講「維持排序陣列+二分搜尋插入」:容易寫對,再主動指出插入要搬動元素、是 O(n)。
- 最佳解是兩個堆積:小的一半放最大堆積、大的一半放最小堆積,兩邊大小最多差一,中位數就在頂端。
和其他主題的關係
時間與空間複雜度(Big O)
| 操作 | 平均 | 最差 |
|---|---|---|
| 每次重新排序:加入+問中位數 | O(n log n) | O(n log n) |
| 排序陣列:加入 找位置 O(log n),但插入要搬開後面的元素 | O(n) | O(n) |
| 排序陣列:問中位數 | O(1) | O(1) |
| 兩個堆積:加入 推進一邊,必要時把一個搬到另一邊 | O(log n) | O(log n) |
| 兩個堆積:問中位數 就是兩個頂端 | O(1) | O(1) |
空間:O(n),三種都得把所有數字存下來
Big O 實測:n 變大時步數怎麼長
數的是:已經有 n 個數字時,再加入一個並回答中位數的工作量(重新排序、排序陣列、兩個堆積分別平均 16、64、512 次)
| Big O | n = 256 | n = 1,024 | n = 4,096 | 成長倍數:實測(理論) | |
|---|---|---|---|---|---|
| 每次重新排序 | O(n log n) | 3,933 | 19,371 | 93,295 | ×24 (×24) |
| 維持排序陣列 | O(n) | 157 | 564 | 2,157 | ×14 (×16) |
| 兩個堆積 | O(log n) | 14.4 | 16.3 | 17.6 | ×1.2 (×1.5) |
和其他做法比
| 每次重新排序 | 維持排序陣列 | 兩個堆積 | |
|---|---|---|---|
| 100 個數字 | 54,532 | 3,015 | 1,025 |
| 250 個數字 | 418,917 | 17,569 | 2,747 |
| 500 個數字 | 1,918,670 | 68,188 | 6,138 |
每一格是整串數字跑完的總工作量(比較加搬動),同一串隨機數字(0–99)給三種解法。重新排序每次都是 n log n,加總起來大約是 n² log n;排序陣列每次插入要把後面一半搬開;兩個堆積每次只動 log n 層。
真實世界裡的它
- 滑動視窗中位數:除了加入還要刪除舊的數字,需要能刪任意元素的結構(例如兩個有序多重集合)。
- 監控系統的 p50、p99 延遲:資料量太大時改用近似的 t-digest 或直方圖,不存每一個值。
- 數字範圍很小(例如 0–100)時,用計數陣列就能 O(範圍) 找中位數。
取捨與陷阱
- Python 的 heapq 只有最小堆積:最大堆積要存負數,拿出來時記得再取負。
- 忘了重新平衡:數字一直偏向某一邊時,兩個堆積的大小會差很多,頂端就不再是中位數。
- 偶數個時用整數除法算平均:(3 + 4) / 2 在某些語言會變成 3。