Модуль 21 ⏱ ~14 мин чтения 📐 1 Mermaid диаграмм

4. Median from Data Stream

«Медиана — самый упрямый из статистических показателей: среднее можно обмануть одним гигантским числом, медиану — никогда».
— Брайан Керниган

Медиана в потоке данных (Median from Data Stream): инвариант двух куч и механика балансировки

Задача непрерывного вычисления медианы потока чисел (Find Median from Data Stream, LeetCode 295) — классический вопрос на стыке алгоритмов и системного мониторинга.

В отличие от среднего арифметического, которое легко вычисляется за $O(1)$ памяти через сумму и счетчик, медиана устойчива к выбросам (Outliers) и аномалиям, что делает ее стандартом для метрик латентности серверов (p50 latency). Однако для точного расчета медианы требуется знать центральный элемент отсортированного ряда. Сортировать весь поток за $O(N \log N)$ при каждом новом числе недопустимо.

Решение заключается в поддержании двух встречно направленных куч (Two Heaps), что обеспечивает вставку за $O(\log N)$ и получение медианы за $O(1)$.

flowchart TD
    classDef process fill:#1e293b,stroke:#38bdf8,stroke-width:1px,color:#f8fafc
    classDef success fill:#064e3b,stroke:#34d399,stroke-width:1px,color:#f8fafc
    classDef error   fill:#7f1d1d,stroke:#f87171,stroke-width:1px,color:#f8fafc
    classDef warning fill:#78350f,stroke:#fbbf24,stroke-width:1px,color:#f8fafc
    classDef entry   fill:#1e3a5f,stroke:#38bdf8,stroke-width:2px,color:#f8fafc

    A["AddNum(num)"]:::entry --> B{"lower пуста или num <= max(lower)?"}:::warning
    B -->|"Да"| C["Push в lower (Max-Heap)"]:::process
    B -->|"Нет"| D["Push в higher (Min-Heap)"]:::process
    C --> E{"Проверка баланса размеров"}:::warning
    D --> E
    E -->|"len(lower) > len(higher) + 1"| F["Переместить корень lower в higher"]:::error
    E -->|"len(higher) > len(lower)"| G["Переместить корень higher в lower"]:::error
    E -->|"Сбалансировано"| H["Инвариант соблюден"]:::success
    F --> H
    G --> H

1. Инвариант двух куч (Two Heaps Invariant)

Числовой ряд мысленно делится строго пополам на две половины:

  1. Нижняя половина (Меньшие числа): организуется как Max-Heap (lower). Корень кучи содержит максимальный из меньших элементов.
  2. Верхняя половина (Большие числа): организуется как Min-Heap (higher). Корень кучи содержит минимальный из больших элементов.

Строгие инварианты структуры:

  1. Инвариант упорядоченности: любой элемент из lower не превышает любого элемента из higher:
    $$\max(\text{lower}) \le \min(\text{higher})$$
  2. Инвариант баланса размеров: размер lower либо равен размеру higher, либо превышает его ровно на $1$:
    $$|\text{lower}| = |\text{higher}| \quad \text{или} \quad |\text{lower}| = |\text{higher}| + 1$$

Вычисление медианы за $O(1)$:

  • Если суммарное количество чисел нечетное ($|\text{lower}| > |\text{higher}|$): медиана равна корню lower.
  • Если суммарное количество чисел четное ($|\text{lower}| = |\text{higher}|$): медиана равна полусумме корней:
    $$\text{median} = \frac{\max(\text{lower}) + \min(\text{higher})}{2.0}$$

2. Идиоматичная реализация на Go

Для исключения аллокаций и накладных расходов интерфейса any реализуем типизированные кучи целых чисел:

Go 
package median

import (
    "container/heap"
    "sync"
)

// maxHeap реализует Max-Heap для нижней половины чисел
type maxHeap []int

func (h maxHeap) Len() int           { return len(h) }
func (h maxHeap) Less(i, j int) bool { return h[i] > h[j] } // Обратный порядок
func (h maxHeap) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }
func (h *maxHeap) Push(x any)        { *h = append(*h, x.(int)) }
func (h *maxHeap) Pop() any {
    old := *h
    n := len(old)
    x := old[n-1]
    *h = old[0 : n-1]
    return x
}

// minHeap реализует Min-Heap для верхней половины чисел
type minHeap []int

func (h minHeap) Len() int           { return len(h) }
func (h minHeap) Less(i, j int) bool { return h[i] < h[j] } // Прямой порядок
func (h minHeap) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }
func (h *minHeap) Push(x any)        { *h = append(*h, x.(int)) }
func (h *minHeap) Pop() any {
    old := *h
    n := len(old)
    x := old[n-1]
    *h = old[0 : n-1]
    return x
}

// MedianFinder координирует балансировку двух куч
type MedianFinder struct {
    mu     sync.RWMutex
    lower  maxHeap // Max-Heap для меньшей половины
    higher minHeap // Min-Heap для большей половины
}

func Constructor() MedianFinder {
    return MedianFinder{
        lower:  make(maxHeap, 0, 64),
        higher: make(minHeap, 0, 64),
    }
}

// AddNum добавляет число в структуру за O(log N)
func (mf *MedianFinder) AddNum(num int) {
    mf.mu.Lock()
    defer mf.mu.Unlock()

    // Шаг 1: Первичная маршрутизация
    if mf.lower.Len() == 0 || num <= mf.lower[0] {
        heap.Push(&mf.lower, num)
    } else {
        heap.Push(&mf.higher, num)
    }

    // Шаг 2: Ребалансировка размеров куч
    // Инвариант: len(lower) == len(higher) или len(lower) == len(higher) + 1
    if mf.lower.Len() > mf.higher.Len()+1 {
        val := heap.Pop(&mf.lower).(int)
        heap.Push(&mf.higher, val)
    } else if mf.higher.Len() > mf.lower.Len() {
        val := heap.Pop(&mf.higher).(int)
        heap.Push(&mf.lower, val)
    }
}

// FindMedian возвращает текущую медиану за O(1)
func (mf *MedianFinder) FindMedian() float64 {
    mf.mu.RLock()
    defer mf.mu.RUnlock()

    if mf.lower.Len() > mf.higher.Len() {
        return float64(mf.lower[0])
    }

    return float64(mf.lower[0]+mf.higher[0]) / 2.0
}

3. Аппаратная механика: кэш-линии и IEEE-754

Локальность данных в срезе кучи

Двоичная куча хранится в непрерывном срезе памяти []int.
Для узла с индексом $i$:

  • Левый потомок: $2i + 1$.
  • Правый потомок: $2i + 2$.
  • Родитель: $\lfloor \frac{i-1}{2} \rfloor$.

Поскольку кучи lower и higher растут одновременно, их размер вдвое меньше суммарного объема потока. Верхние уровни дерева кучи (первые 4–5 поколений, до 31 элемента) непрерывно удерживаются в кэше L1/L2 процессора, делая первые шаги операций siftUp и siftDown практически бесплатными.

4. Вопросы для собеседования

Резюме

  1. Алгоритм Two Heaps разделяет поток на две половины: Max-Heap для меньшей части и Min-Heap для большей.
  2. Поддержание баланса размеров гарантирует доступ к медиане ровно за $O(1)$.
  3. Любое добавление элемента требует не более одного шага ребалансировки за $O(\log N)$.
  4. При вычислении полусуммы корней четного потока необходимо избегать целочисленного переполнения и потери дробной части.

В следующей статье мы разберем расчет медианы в бесконечном потоке с фиксированным расходом памяти $O(1)$ с помощью алгоритма $P^2$: 5. Find Median in Infinite Stream.