Распределённые системы
Шардирование
Цели урока
- Объяснить зачем шардировать и чем шардирование отличается от репликации
- Сравнить hash, range и directory стратегии и назвать trade-off каждой
- Выбрать правильный sharding key для конкретного access pattern
- Описать стратегии rebalancing и почему fixed partitions лучше naive hash % N
Предварительные знания
- Понимание репликации (leader-follower, eventual consistency)
- Базовое знание хэш-функций (MD5, consistent hashing - понятие)
- CAP theorem: почему при partition нужно выбирать между C и A
Instagram в 2012 году: 30 млн пользователей, 150 млн фотографий, 13 инженеров - и PostgreSQL, вручную разбитый на шарды. Шардирование было не роскошью, а условием выживания.
- **Cassandra (Netflix)** - consistent hash sharding, каждый шард реплицируется x3, хранит терабайты пользовательских данных
- **Kafka** - партиции (шарды) = единица параллелизма, consumer group читает каждую партицию независимо
- **Redis Cluster** - 16384 hash slots распределены по узлам, manual resharding
- **CockroachDB** - auto-split при превышении 512 MB, Raft per range, глобальная ACID
- **X (бывш. Twitter)** - fan-out-on-write: при публикации твит шардируется в timeline каждого фолловера (денормализация вместо scatter-gather)
Consistent Hashing и рождение distributed storage
В 1997 году Karger, Lehman, Leighton, Panigrahy, Levine и Lewin опубликовали статью 'Consistent Hashing and Random Trees' на STOC. Задача была простой: как минимизировать количество переносов данных при добавлении/удалении сервера в distributed cache. Решение через кольцо хэшей оказалось настолько универсальным, что 27 лет спустя его используют Cassandra, DynamoDB, Riak, Akamai CDN и тысячи других систем без существенных изменений.
Что такое шардирование и зачем оно нужно
**2012 год. Instagram продаётся Facebook за 1 млрд долларов. На момент сделки в системе 30 млн пользователей и 150 млн фотографий - и всё это на PostgreSQL, разбитом на шарды вручную командой из 13 инженеров.** Один инстанс БД к тому времени уже не мог принять нагрузку - шардирование стало единственным выходом.
**Шардирование** (partitioning) - разделение данных на части (шарды), каждая из которых хранится на отдельном сервере. Sharding и partitioning - синонимы. MongoDB называет это sharding, Kafka - partitioning, PostgreSQL - table partitioning.
| Сценарий | Один сервер | 10 шардов |
|---|---|---|
| Объём данных | Ограничен железом (10-50 TB) | 10x ёмкость без апгрейда |
| Запросы в секунду | Bottleneck ~10-50K QPS | Линейный рост с числом шардов |
| Отказоустойчивость | Single point of failure | Отказ шарда = потеря части данных |
| Geo-latency | Одна локация | Данные ближе к пользователю |
Шардирование не заменяет репликацию - каждый шард обычно ещё и реплицируется для отказоустойчивости. Это два независимых измерения масштабирования: репликация решает проблему доступности и read throughput, шардирование - объёма данных и write throughput.
| Компонент | Роль | Что хранит |
|---|---|---|
| Client | Отправляет запрос | - |
| Router / Coordinator | Маршрутизация по диапазону user_id | - |
| Shard 1 | Данные пользователей A-F | 1/3 нагрузки |
| Shard 2 | Данные пользователей G-M | 1/3 нагрузки |
| Shard 3 | Данные пользователей N-Z | 1/3 нагрузки |
**Не стоит шардировать преждевременно.** Шардирование - не серебряная пуля, а серьёзный архитектурный сдвиг: joins между шардами невозможны, транзакции становятся распределёнными (2PC), мониторинг усложняется. Сначала реплики, кэш, индексы - и только потом шардирование.
Шардирование и репликация - это одно и то же, просто разные названия
Это два разных механизма: репликация дублирует данные для надёжности и read scale, шардирование разбивает данные для write scale и ёмкости
В production-системах используют оба: каждый шард обычно имеет 2-3 реплики. Cassandra, MongoDB и HBase работают именно так.
Стратегии шардирования: Hash, Range, Directory
**Три стратегии шардирования - три разных ответа на вопрос: "на каком шарде искать данные?"** Выбор определяет всё: от query performance до сложности rebalancing при изменении числа шардов.
| Стратегия | Как работает | Плюсы | Минусы |
|---|---|---|---|
| Hash | shard = hash(key) % N | Равномерное распределение | Resharding при изменении N - все данные переезжают |
| Range | A-M на shard1, N-Z на shard2 | Эффективны range queries | Hot spots при skewed distribution |
| Directory | Отдельный lookup-сервис знает где что | Максимальная гибкость | SPOF, дополнительный latency на lookup |
**Consistent Hashing** решает проблему resharding у hash-стратегии: при добавлении шарда переезжает только 1/N данных, а не (N-1)/N. Именно поэтому Cassandra, DynamoDB и Redis Cluster используют consistent hashing. Подробно - в отдельном уроке.
Range sharding: логи по времени
ClickHouse хранит логи с range partitioning по дате: shard_2024_01 хранит январь, shard_2024_02 - февраль и т.д. Запросы типа "логи за последнюю неделю" идут на 1-2 шарда. Старые шарды переходят в режим read-only и могут быть перенесены на дешёвые HDD. Новые записи всегда идут на текущий шард - это создаёт hot spot, который компенсируется тем, что текущий шард самый мощный.
**Directory-based sharding** гибка, но lookup-сервис становится SPOF (single point of failure) и узким местом по latency. Если lookup недоступен - вся система встаёт. Facebook использовал эту стратегию для MySQL и решал проблему через репликацию самого lookup-слоя.
Команда добавляет 4-й шард к системе с 3 шардами (простой hash % N). Какой процент данных нужно перенести?
Выбор sharding key и cross-shard операции
**Sharding key - самое важное архитектурное решение при шардировании.** Неправильный выбор приводит к hot spots (один шард под 90% нагрузки при простаивающих остальных) или scatter-gather (каждый запрос идёт на все шарды). Оба сценария уничтожают преимущества шардирования.
| Критерий | Что это означает | Пример |
|---|---|---|
| High cardinality | Много уникальных значений для равномерного распределения | user_id хорошо, status (3 значения) плохо |
| Access pattern | Запросы должны включать sharding key, иначе scatter-gather | Если ищут по email - ключ email, не user_id |
| Avoid hot spots | Не создаёт перекосы в нагрузке | Timestamp плох - все новые записи на один шард |
| Immutable | Изменение ключа = перемещение данных между шардами | user_id лучше username (username меняется) |
Scatter-Gather: цена запроса без sharding key
Запрос: SELECT * FROM orders WHERE total > 50000 - не содержит sharding key (user_id). Без ключа router рассылает запрос на все N шардов (scatter), ждёт ответа от всех, сливает результаты (gather). Latency = max(latency всех шардов). Нагрузка = N x стоимость одного запроса. С sharding key: WHERE user_id = 42 AND total > 50000 - идёт на один шард. Latency минимальна.
| Данные | Хороший ключ | Плохой ключ |
|---|---|---|
| Пользователи | user_id | country (skew - США 40% юзеров) |
| Заказы | user_id | status (3 значения = 3 шарда) |
| Сообщения | conversation_id | timestamp (hot spot на текущем) |
| Логи | timestamp + source | level (3-5 значений) |
**Cross-shard транзакции** требуют 2PC (two-phase commit) - дорого и хрупко. Данные проектируются так, чтобы транзакция оставалась внутри одного шарда. Правило: сущности, которые изменяются вместе, должны жить на одном шарде (co-location).
**Secondary indexes** при шардировании: Local Index - индекс только на одном шарде, writes атомарны, но query без sharding key = scatter-gather. Global Index - один индекс знает где всё, query эффективен, но writes требуют обновления глобального индекса (eventual consistency). DynamoDB Global Secondary Index работает именно так.
Можно выбрать любой атрибут как sharding key и добавить индексы для остальных запросов
Rebalancing: как добавлять и удалять шарды
**Данные растут, шарды переполняются, серверы ломаются - нужно перераспределить данные без простоя.** Rebalancing - самая болезненная операция при шардировании. Instagram провёл несколько таких миграций онлайн, пока сервис работал под нагрузкой 5 млн запросов в минуту.
| Стратегия | Принцип | Преимущество |
|---|---|---|
| Fixed partitions | Много маленьких партиций (1000), распределены по серверам. Новый сервер - переносим часть партиций. | Минимум перемещения данных (~10% при добавлении сервера) |
| Dynamic partitioning | Шард перегружен - делим пополам. Пустой - сливаем с соседом. | Автоматический баланс, нет ручного управления. HBase, MongoDB |
| Partition by node | Фиксированное число партиций на узел. Новый узел - крадёт партиции у остальных. | Предсказуемый размер партиций. Cassandra |
**Consistent Hashing** реализует fixed partitions через виртуальные ноды на кольце хэшей. Cassandra: каждый физический узел владеет несколькими диапазонами кольца (vnode). Добавление узла - перенос части vnodes.
Шардирование в реальных системах
MongoDB: автоматический rebalancer перемещает chunks (64 MB по умолчанию) между шардами когда разница >8 chunks. Cassandra: consistent hash + vnodes, auto-rebalance при добавлении/удалении. Redis Cluster: 16384 hash slots, ручной resharding через redis-cli --cluster. CockroachDB: range-based, автоматическое разбиение при превышении 512 MB, Raft per range.
- **Anti-pattern: шардировать слишком рано** - overhead без необходимости. Сначала реплики и кэш.
- **Anti-pattern: плохой sharding key** - hot spots или scatter-gather на каждый запрос.
- **Anti-pattern: cross-shard joins** - медленно и ненадёжно. Денормализация или переосмысление паттерна доступа.
- **Anti-pattern: игнорировать locality** - связанные данные на разных шардах = много network calls.
- **Anti-pattern: фиксированное число шардов** - трудно масштабировать. Consistent hashing или много маленьких партиций.
Шардинг по hash(user_id) автоматически даёт равномерную нагрузку
Hash-шардинг балансирует объём данных, но не нагрузку: hot keys (топ-инфлюенсеры, viral posts) создают перегруз одного шарда даже при идеальной хэш-функции
Реальные распределения активности follow power-law (Zipf): 1% пользователей генерирует 50%+ запросов. Решение - composite key (user_id, bucket_id) с дополнительной фрагментацией hot keys, либо отдельный путь через cache/replicas для hot tenants. Production-системы (Discord, Slack) явно отделяют large servers от small servers в разные шард-стратегии.
Связь с предыдущим
Репликация решает доступность и read-throughput, но не write-bottleneck. Когда working set не помещается в RAM одного сервера, остаётся только горизонтальное разделение данных.
- Replication — масштабирует чтения и даёт HA, но master по записи один
- Partitioning strategies — range, hash, directory - три способа маршрутизации запроса к нужному шарду за O(1)
- Consistent hashing — продолжение темы в следующем уроке: добавление узла без миграции всего датасета
Итоги
- Выбор sharding key определяет всё: правильный ключ даёт равномерное распределение и локальность для типовых запросов, неверный приводит к hot-shards и scatter-gather на каждый запрос
- Range-партиции дают эффективные диапазонные запросы, но требуют адаптивного splitting (HBase/CockroachDB), иначе появляются hot ranges в конце диапазона при монотонных ключах
- Hash-партиции равномерны, но убивают range-запросы: scatter-gather по всем шардам становится дефолтным паттерном чтения
- Directory-based sharding (lookup table) даёт максимальную гибкость, но добавляет SPOF в виде directory-сервиса и требует кэширования на клиентах
- Fixed-partitions подход (например, 16384 слота в Redis Cluster) разрывает связь между числом шардов и числом узлов: rebalancing переносит только нужные chunks, не пересчитывая хэши
Вопросы для размышления
- Система хранения сообщений мессенджера: по какому ключу шардировать: user_id отправителя, user_id получателя или conversation_id? Какие запросы станут эффективными, а какие потребуют scatter-gather?
Связанные уроки
- dist-15-consistent-hashing — Consistent hashing - основа hash-шардирования
- ds-04-consistent-hashing — Минимизирует перенос ключей при rebalance
- dist-11-replication — Каждый шард обычно реплицируется независимо
- ds-05-replication — Репликация и шардирование - ортогональные оси масштабирования
- dist-07-2pc — Cross-shard транзакции требуют 2PC
- dist-12-consistency — Шардирование меняет consistency-модель запросов
- db-23-sharding