Спроектируйте счётчик просмотров для видео с миллионами просмотров в минуту.

Требования: отображать число просмотров (и лайков) на странице видео, учитывать каждый просмотр, не считать накрутки, выдерживать вирусные пики. Обычно допустимо: значение отстаёт на секунды, «1,2 млн» вместо точного числа.

Оценка: 50 000 просмотров в секунду в среднем, пики ×5; один вирусный ролик может давать 10% всех просмотров — 25 000 инкрементов в секунду одного ключа.

Почему не прямой UPDATE: каждый инкремент берёт блокировку строки; одна строка выдерживает от сотен до пары тысяч обновлений в секунду, и очередь на блокировку растёт. Плюс WAL и репликация на каждое событие.

Конвейер:

1. клиент или сервис раздачи отправляет событие просмотра (видео, пользователь, время, длительность) → балансировщик → приёмник, который сразу кладёт в Kafka (партиции по video_id) и отвечает;
2. агрегатор читает партиции, копит счётчики в памяти (video_id → +N) и раз в 1–5 секунд сбрасывает дельты пачкой: UPDATE ... SET views = views + delta — одна запись вместо тысяч;
3. смещение Kafka фиксируется после сброса — при падении агрегатора события перечитываются (возможен двойной счёт пачки — либо смириться, либо идемпотентный сброс с номером пачки);
4. чтение — из кэша (Redis), обновляемого при сбросе; страница показывает значение с задержкой в секунды.

Горячие ключи: партиция с вирусным видео перегружает одного агрегатора — шардировать ключ (video#0..15) или предагрегировать на приёмниках. В Redis — шардированные счётчики и сумма при чтении.

Уникальные зрители — HyperLogLog на видео и день (12 КБ на ключ), объединение для периодов.

Накрутки: просмотр засчитывается при просмотре дольше N секунд; дедупликация по пользователю/устройству в окне (Redis SET NX с TTL или фильтр Блума); антифрод-анализ потом может «откатить» часть просмотров — поэтому публичное число иногда замирает (как у YouTube на 301).

Аналитика (просмотры по часам, регионам) — те же события в ClickHouse, отдельно от счётчика для страницы.

from collections import Counter
import time

pending, last_flush = Counter(), time.monotonic()

for event in consumer:                         # Kafka, партиции по video_id
    pending[event.video_id] += 1
    if time.monotonic() - last_flush > 2:
        db.execute_batch("UPDATE videos SET views = views + %s WHERE id = %s",
                         [(n, vid) for vid, n in pending.items()])
        consumer.commit()                      # смещение — после записи
        pending.clear(); last_flush = time.monotonic()