Кратко и по делу — предлагаемая архитектура + обоснования и численные оценки.
1) Выбор потоковой vs батч
- Рекомендация: потоковая обработка (streaming) для near‑real‑time метрик; батч (daily) — проще, но даёт задержку в часы.
- Причина: $10^9$ событий/день → средняя нагрузка $\frac{10^9}{86400}\approx 11574$ событий/с, стриминг легко масштабируется и даёт быстрый отклик и SLA на свежесть.
2) Алгоритм подсчёта уникальных (как минимизировать трафик и память)
- Использовать приближённый наборный алгоритм с поддержкой гармоничного объединения, например HyperLogLog (HLL или HyperLogLog++).
- Ошибка HLL: приблизительно $\text{err}\approx \frac{1.04}{\sqrt{m}}$. Для ошибки $\approx 1\%$ нужно $m\approx {\left(\frac{1.04}{0.01}\right)}^2\approx 10816$. Практически выбираем $m=2^{14}=16384$ (ошибка $\approx \frac{1.04}{128}\approx 0.81\%$).
- Память одного HLL при $m=16384$ ≈ $16384\times 6$ бит ≈ $98$ KB (порядок десятков килобайт).
3) Топология сбора данных (иерархическое агрегационное шардирование)
- На каждом edge/региональном сервере:
- Хешировать уникальный идентификатор посетителя (user_id) и апдейтить локальный HLL за дневное окно.
- Периодически (например, каждые $T=1$–$5$ минут) отправлять компактный HLL (или дельту) к региональному агрегатору. Это снижает сетевой трафик по сравнению с отправкой каждого события.
- Региональный уровень:
- Принимает HLL от множества источников, объединяет (union) — это ассоциативная, коммутативная и идемпотентная операция.
- Периодически пушит агрегированные HLL в центральный уровень (или в Kafka топик).
- Центральный уровень:
- Объединяет региональные HLL → итоговый дневной HLL → вычисляет estimate уникальных.
- Шардирование:
- Для точной дедупликации можно альтернативно шардировать поток по $h(user\_id)\mathrm{mod}N$, где $N$ — число партиций/шардов. Тогда все события для одного user_id попадают в один шард и можно держать точные структуры (или HLL per shard). Выбор $N$ — по пиковому QPS и state size.
4) Консистентность и семантика доставки
- Очень важно: HLL merge — идемпотентен, поэтому at‑least‑once семантика приемлема (повторы не увеличивают оценку). Это упрощает отказоустойчивость.
- Для записей в конечное хранилище (сохранять snapshot HLL / estimate) — обеспечить как минимум exactly‑once/transactional через потоковую платформу (Kafka transactions / Flink exactly‑once sinks) если критична отсутствие дубликатов в метаданных.
- В целом: выбрать eventual/near‑real consistency (обновления каждые $T$ минут), strong‑consistency не требуется для агрегированной дневной метрики.
5) Отказоустойчивость / долговечность
- Ingress: использовать распределённый журнал (Kafka) с replication factor $=3$, retention ≥ сутки + запас.
- Stream processing: использовать Flink/Beam/Spark Structured Streaming с state backend (RocksDB), периодическими checkpoint'ами в распределённое хранилище (S3/HDFS). Это обеспечивает восстановление state и минимальную потерю прогресса.
- Репликация состояния: хранить резервные snapshot'ы HLL для каждого шардa.
- Региональные агрегаторы: активный/пассивный или несколько реплик, при падении восстановление из Kafka + checkpoint.
6) Как минимизировать задержку и сетевой трафик
- Отправлять не события, а HLL/скетчи: вместо $E$ байт/событие → $S$ байт/скетч каждые $T$ секунд. Пример: 1000 источников, $S\approx 100$ KB, $T=60$ с → трафик = $1000\times 100\text{KB}/60\approx 1.67\text{MB/s}$ ≈ $144$ MB/мин → за день ≈ $207$ GB? (проверьте параметры). Можно уменьшить $S$ и/или увеличить $T$.
- Более реалистный расчёт: для 1000 источников, HLL≈100 KB, $T=5$ мин → трафик/день ≈ $1000\times 100\text{KB}\times (1440/5)\approx 28.8$ GB/день.
- Сжатие перед отправкой (gzip) + бинарный формат уменьшат трафик в ~3×.
- Delta/инкрементальные скетчи: посылать только изменённые регистры (если реализация позволяет) → уменьшение трафика.
- Локальная фильтрация: на боковых серверах можно держать небольшой LRU кеш недавних visitor_id для снижения частых повторов (если пользователи шлют много событий подряд).
- Использование CDN/региональных агрегаторов: снизит межконтинентальный трафик.
7) Операционные детали и окна времени
- Окно — календарный день (UTC или локальный): реализовать ротацию state в midnight boundary, использовать watermarks для событий с задержкой.
- Поздние данные: держать буфер (например, 1–2 часа) и применять корректировки через повторное объединение HLLs; HLL позволяет объединять новые данные без переписывания старых.
- Мониторинг: отслеживать size of HLL registers, traffic, latency, divergence между региональными и центральными оценками.
8) Пример pipeline (конкретные компоненты)
- Edge producers → Kafka topic (partitioned by hash(user_id) или отправляют локальные HLLs) → Stream processor (Flink/Beam) с state per day (HLL per shard) → Periodic snapshots → Object store (S3) + OLAP DB (ClickHouse/BigQuery) для аналитики → API/BI.
9) Числовые сводки (оценка)
- Событий/с: $\frac{10^9}{86400}\approx 11574$.
- HLL размер: при $m=16384$ ≈ $98$ KB, ошибка $\approx \frac{1.04}{\sqrt{16384}}=\frac{1.04}{128}\approx 0.0081$ (0.81%).
- Трафик при 1000 источниках, $T=5$ мин, HLL $100$ KB: $\approx 28.8$ GB/день (до сжатия).
10) Вывод / рекомендации
- Используйте streaming (Flink/Beam) + Kafka; подсчёт через HLL (HyperLogLog++) с региональной и централизованной агрегацией; шардируйте по hash(user_id) для балансировки; at‑least‑once приемлемо благодаря идемпотентности HLL; checkpointing + Kafka replication для отказоустойчивости; отправляйте скетчи периодически, применяйте сжатие и дельты для минимизации трафика; ротация state по дневным окнам и мониторинг.
Если нужно, могу дать конкретную конфигурацию (N партиций, параметр m для HLL, T для отправки) под ваши целевые ограничения по точности/латентности/трафику — укажите допустимую погрешность и желаемую частоту обновлений.