Как найти самые популярные элементы в бесконечном потоке событий?

Задача. Поток из миллиардов событий (запросы, клики, ключи кэша), различных элементов — сотни миллионов. Хранить счётчик на каждый — слишком много памяти. Нужна частота популярных и топ-K.

Count-min sketch. Матрица из d строк по w счётчиков, у каждой строки своя хеш-функция.

• событие x — в каждой строке i увеличить счётчик hᵢ(x) mod w;
• оценка частоты x — минимум из d счётчиков.

Каждый счётчик — это частота x плюс частоты всех, кто попал в ту же ячейку, поэтому оценка никогда не меньше истины. Минимум по строкам выбирает ячейку с наименьшим «мусором». Гарантия: с вероятностью 1 − δ ошибка не больше ε·N, где w = e/ε, d = ln(1/δ). Например, w = 2000, d = 5 — ошибка до 0,14% потока с вероятностью 99%, памяти 10 000 счётчиков. Для редких элементов оценка бесполезна, для частых — точна.

Топ-K. Рядом со скетчем держим мин-кучу на K элементов: после обновления оцениваем x, и если оценка больше минимума кучи — заменяем. Итог — K самых частых с приближёнными частотами.

Space-Saving — K счётчиков; новый элемент при заполнении вытесняет элемент с минимальным счётчиком и наследует его значение + 1. Простая, точная для «тяжёлых» элементов, используется во многих системах мониторинга. Misra-Gries — похожий детерминированный вариант.

Окна времени. «Топ за последний час» — хранить скетчи по минутам, складывать последние 60 (count-min складываются поячеечно) и выкидывать старые. Или затухание: счётчики периодически умножать на коэффициент.

Распределённо. Каждый узел считает свой топ, координатор сливает. Ловушка: элемент может быть 101-м на всех узлах и первым в сумме. Поэтому узлы отдают топ с запасом (например, 10·K) или скетчи целиком — скетчи складываются. Точный топ — пакетно: MapReduce, ClickHouse topK (приближённый) или точная агрегация по ночам.

Применения: популярные запросы и тренды, горячие ключи кэша (W-TinyLFU в Caffeine использует count-min как частотный фильтр), обнаружение DDoS по источникам, тяжёлые пользователи API.

import random

class CountMin:
    def __init__(self, w=2000, d=5, seed=1):
        rnd = random.Random(seed)
        self.w, self.rows = w, [[0] * w for _ in range(d)]
        self.salts = [rnd.getrandbits(64) for _ in range(d)]

    def add(self, x, c=1):
        for row, s in zip(self.rows, self.salts):
            row[hash((s, x)) % self.w] += c

    def estimate(self, x):
        return min(row[hash((s, x)) % self.w] for row, s in zip(self.rows, self.salts))