Кратко: цель — надежно и эффективно посчитать частоты слов в корпусе размера ~$\text{PB}$. Ниже — проект решения: партиционирование, борьба со skew (горячие ключи), отказоустойчивость, гарантии консистентности и механизмы перезапуска; затем сравнение архитектур и рекомендации.
1) Партиционирование (основной поток)
- Разбиение входа: читаем корпус как набор больших сплитов (например, HDFS-блоки) размера $\sim 128\text{MB}$ (или другой оптимальный блок) и назначаем каждому map-задачу один или несколько сплитов.
- Партиция по ключу: каждому слову назначается партиция через хеш-функцию: ключ reducer = $\text{hash(word)}\mathrm{mod}R$, где $R$ — число редьюсеров. Ожидаемая загрузка на редьюсер $\approx \frac{S}{R}$, где $S$ — объём данных.
- Map-side комбинирование: на карте локально аггрегируем (combiner) частоты до отправки по сети, чтобы сократить shuffle-объём.
- Двухуровневая агрегация (опционально): map → промежуточные агрегаты → промежуточный shuffle между небольшим числом «pre-aggregator» воркеров → финальные редьюсеры. Это полезно при больших объёмах.
2) Обработка skew-словарей (горячих ключей)
- Обнаружение горячих ключей: предварительная выборка/sampling первых $s$ байт или случайных фрагментов, вычисление top-$K$ слов (например, $K=1000$) и оценка ожидаемой частоты $C_w$ для слова $w$.
- Шардирование горячих ключей: для слова $w$ с ожидаемой нагрузкой > целевой нагрузки $T$ делим его на $m$ шардов, где
$$m\approx \lceil \frac{C_w}{T}\rceil .$$ Map-выходы для горячего слова эмитятся как ключи $(w,shard\_id)$. Затем специальный финализатор суммирует все шард-части в окончательный счет.
- Альтернативы/дополнения:
- dynamic work stealing: если редьюсер перегружен, часть шарда перекидывается на свободные узлы;
- count-min sketch на этапе map для приблизительных частот (уменьшает память, даёт быстрый фильтр на heavy hitters);
- выделенные «hot-key» воркеры для очень частых токенов (например, стоп-слова), которые принимают поток и аггрегируют в памяти с периодическим сбросом.
- Замечание: шардирование сохраняет корректность, потому что операция суммирования ассоциативна и коммутативна.
3) Обеспечение отказоустойчивости
- Batch (MapReduce/Spark):
- детерминированные map задачи + локальные комбиниры; при падении — перезапуск map/reduce по входным сплитам;
- промежуточные результаты пишутся в устойчивое хранилище (HDFS/объектное хранилище) как временные файлы; финальная отдача — атомарный rename/commit;
- мастер отслеживает попытки задач (attempt IDs) и предотвращает «параллельное» финализирование.
- Stream (Flink/Spark Structured Streaming):
- сохраняем оффсеты входного лога (Kafka) и делаем регулярные контрольные точки состояния (checkpoints); при падении восстанавливаемся от последнего checkpoint и повторно читаем лог;
- state backend (RocksDB/remote storage) для больших состояний.
- Serverless:
- функции статeless, все промежуточные результаты — в durable store (S3, DynamoDB); сохранить idempotent writes (ключи с попыткой/UUID) — чтобы повторный вызов не давал дубликатов.
- Для всех подходов: используем таймауты, лимиты повторов, экспоненциальный бэкофф, трекинг метрик.
4) Гарантии консистентности
- Модели:
- at-most-once — может терять события (не подходит, если важны точные счёты);
- at-least-once — события могут повторяться → риск двойного подсчёта;
- exactly-once — идеал: каждое вхождение засчитывается ровно один раз.
- Для WordCount (ассоциативная и коммутативная операция «сумма») можно получить практическое exactly-once с комбинацией:
- детерминированная партиция и idempotent final-reduce (т.е. для каждой (слово, shard) результат записывается с уникальным ключом и переапдейты делаются как overwrite или атомарная сумма с контрольной меткой);
- транзакционный/атомарный commit финальных файлов (rename);
- в стриме — transactional sinks (поддержка commit protocol) или use two-phase commit/atomic writes.
- Часто разумная стратегия: at-least-once + idempotent merge/пересчёт → фактическая корректность итогов.
5) Механизм перезапуска задач и согласованность результатов
- В batch: мастер хранит metadata о сплитах и попытках; при падении нода помечается, её сплиты возвращаются в очередь; map задачи детерминированы — можно перезапустить; редьюсеры перезапускаются, если их временные файлы неполные; финальные файлы публикуются только после успешного commit.
- В stream: checkpointing state + оффсеты; при рестарте воркер поднимает state и продолжает с последнего checkpoint; нужны exactly-once sinks или идемпотентные записи.
- В serverless: оркестратор (step functions, workflow) хранит прогресс, при сбое реинициализирует функции; чтобы избежать дубликатов — используем idempotent keys или хранение попыток.
6) Архитектуры и компромиссы
- MapReduce / Batch (Hadoop, Spark batch)
- Плюсы: простота модели, масштабируемость до $\text{PB}$, хорошие отказоустойчивые механизмы (перезапуск задач), низкие требования к постоянному обслуживанию.
- Минусы: высокая латентность (время прохождения job может быть часами), большой сетевой shuffle и IO; ручная настройка под skew.
- Лучшее применение: однократный/периодический подсчёт по большим корпусам.
- Stream-processing (Flink, Spark Structured Streaming)
- Плюсы: низкая латентность, инкрементальные результаты, встроенные механизмы checkpoint/restore, возможность поддерживать большие состояния (RocksDB).
- Минусы: сложнее справляться с экстремальным skew (hot keys требуют шардирования или adaptive routing), операционная сложность, потенциал для backpressure; state management при $\text{PB}$ данных требует продуманной архитектуры.
- Лучшее применение: постоянный поток документов, где нужны near‑real-time обновления счётчиков.
- Serverless (FaaS + облачное хранилище)
- Плюсы: эластичность, простота управления, оплата по использованию.
- Минусы: лимиты времени/памяти на функцию, высокая латентность при большом shuffle, тяжело хранить и быстро объединять большие состояния, стоимость может вырасти при $\text{PB}$-масштабе; координация большого числа функций — сложна.
- Лучшее применение: триггерная обработка, промежуточная агрегация, гибридные подходы (serverless для предобработки, batch для финальной агрегации).
7) Рекомендация для корпуса в $\text{PB}$ - Основной вариант: batch-ориентированная распределённая обработка (MapReduce / Spark) с:
- map-side combine, двухуровневая агрегация,
- предварительным sampling для выявления heavy hitters,
- шардированием горячих ключей по формуле $m\approx \lceil C_w/T\rceil$,
- хранением промежуточных файлов в отказоустойчивом хранилище и атомарным commit финала,
- мониторингом и адаптивным увеличением $R$ / выделением ресурсов на hot-keys.
- Для онлайна/инкремента: гибрид — периодический batch пересчёт + стриминговое приближение для последних событий (streaming for deltas, batch for full re-compute).
- Для экстремального skew комбинируйте: heavy-hitter таблицу (список очень частых слов) обслуживать отдельными выделенными сервисами/ключ-аггрегаторами; все остальные — обычная партиция.
8) Операционные практики
- Логирование и метрики: per-partition size, shuffle bytes, task latency, hotspot detection.
- Тестирование на репликах: synthetic tests с искусственно введёнными hot keys.
- Автоматический ресайзинг: менять $R$ и число шардов для hot-keys по метрикам.
- Версионирование вывода и возможность отката.
Если нужно — могу дать конкретную конфигурацию параметров (как выбирать $R$, $T$, $m$, размер сплита) для ваших входных данных: скажите ожидаемый объём корпуса в $\text{PB}$ и среднюю плотность токенов на байт.