key-value хранилищe (redis)
Key-value store, также называемое key-value базой данных, — это нереляционная база данных. Каждый уникальный идентификатор хранится как ключ вместе со связанным с ним значением. Такая пара данных и называется парой «ключ-значение».
В паре key-value ключ должен быть уникальным, а доступ к значению осуществляется через этот ключ. Ключи могут быть обычным текстом или хешированными значениями. С точки зрения производительности короткие ключи работают лучше. Как могут выглядеть ключи? Вот несколько примеров:
-
Текстовый ключ:
last_logged_in_at -
Хешированный ключ:
253DDEC4
Значением в паре key-value могут быть строки, списки, объекты и т. д. В большинстве key-value хранилищ, например Amazon Dynamo [1], Memcached [2], Redis [3], значение обычно рассматривается как непрозрачный объект.
Вот пример фрагмента данных в key-value хранилище:
| key | value |
|---|---|
| 145 | john |
| 147 | bob |
| 160 | julia |
Таблица 1
В этой главе нужно спроектировать key-value store, поддерживающее следующие операции:
-
put(key, value)// вставитьvalue, связанное сkey -
get(key)// получитьvalue, связанное сkey
Понять задачу и определить рамки дизайна
Идеального дизайна не существует. Каждый вариант достигает определенного баланса между чтением, записью и потреблением памяти. Еще один компромисс приходится делать между consistency и availability. В этой главе мы проектируем key-value store со следующими характеристиками:
-
Размер пары key-value небольшой: менее 10 КБ.
-
Возможность хранить большие объемы данных.
-
Высокая доступность: система должна быстро отвечать даже при сбоях.
-
Высокая масштабируемость: систему можно масштабировать для поддержки больших объемов данных.
-
Автоматическое масштабирование: добавление и удаление серверов должно происходить автоматически в зависимости от трафика.
-
Настраиваемая консистентность.
-
Низкая задержка.
Key-value store на одном сервере
Сделать key-value store, работающее на одном сервере, несложно. Самый интуитивный вариант — хранить пары key-value в хеш-таблице, полностью размещенной в памяти. Хотя доступ к памяти быстрый, уместить все данные в памяти может быть невозможно из-за ограничений по объему. Чтобы поместить больше данных на одном сервере, можно использовать две оптимизации:
-
Сжатие данных
-
Хранение в памяти только часто используемых данных, а остального — на диске
Но даже с этими оптимизациями один сервер довольно быстро упрется в свой предел. Для работы с большими объемами данных нужно распределенное key-value хранилище.
Распределенное key-value хранилище
Распределенное key-value хранилище также называют distributed hash table: пары key-value распределяются по многим серверам. При проектировании распределенной системы важно понимать теорему CAP (Consistency, Availability, Partition Tolerance).
Теорема CAP
Согласно теореме CAP, распределенная система не может одновременно обеспечивать больше двух из трех гарантий: consistency, availability и partition tolerance. Сначала зафиксируем определения.
Consistency: все клиенты видят одни и те же данные в один и тот же момент времени независимо от того, к какому узлу они подключаются.
Availability: любой клиент, запросивший данные, получает ответ, даже если часть узлов недоступна.
Partition tolerance: partition означает разрыв связи между двумя узлами. Partition tolerance означает, что система продолжает работать, несмотря на сетевые разделения.
Теорема CAP говорит, что из этих трех свойств нужно пожертвовать одним, чтобы получить два остальных, как показано на рисунке 1.
Рисунок 1
Сегодня key-value хранилища обычно классифицируют по тем двум CAP-характеристикам, которые они поддерживают:
CP (consistency + partition tolerance): CP key-value store поддерживает consistency и partition tolerance, жертвуя availability.
AP (availability + partition tolerance): AP key-value store поддерживает availability и partition tolerance, жертвуя consistency.
CA (consistency + availability): CA key-value store поддерживает consistency и availability, жертвуя partition tolerance. Но поскольку сетевые сбои неизбежны, реальная распределенная система должна терпимо относиться к network partition. Поэтому CA-системы в реальном мире не существуют.
Выше была в основном теория. Чтобы стало понятнее, рассмотрим конкретный пример. В распределенных системах данные обычно реплицируются несколько раз. Предположим, данные реплицируются на трех репликах: n1, n2 и n3, как показано на рисунке 2.
Идеальная ситуация
В идеальном мире сетевых разделений не происходит. Данные, записанные на n1, автоматически реплицируются на n2 и n3. В таком случае достигаются и consistency, и availability.
Рисунок 2
Реальные распределенные системы
В распределенной системе partitions неизбежны, а когда они происходят, приходится выбирать между consistency и availability. На рисунке 3 узел n3 выходит из строя и больше не может взаимодействовать с n1 и n2. Если клиенты пишут данные в n1 или n2, эти данные не могут быть распространены на n3. Если же данные были записаны в n3, но еще не успели реплицироваться в n1 и n2, то n1 и n2 будут содержать устаревшие данные.
Рисунок 3
Если выбрать consistency вместо availability (CP-система), то нужно заблокировать все операции записи в n1 и n2, чтобы избежать рассинхронизации данных между тремя серверами. Это делает систему недоступной. Банковские системы часто предъявляют крайне высокие требования к consistency. Например, для банка критически важно показывать самый актуальный баланс. Если из-за network partition возникает inconsistency, банковская система скорее вернет ошибку, чем покажет неверный результат.
Если же выбрать availability вместо consistency (AP-система), система продолжит принимать чтения, хотя и может возвращать устаревшие данные. Для записей n1 и n2 продолжат принимать новые значения, а после восстановления network partition эти данные будут синхронизированы с n3.
Выбор подходящего CAP-профиля под конкретный сценарий использования — важный шаг при построении распределенного key-value хранилища. Это стоит обсудить с интервьюером и дальше проектировать систему соответствующим образом.
Компоненты системы
В этом разделе разберем основные компоненты и техники, используемые при построении key-value store:
-
Разбиение данных
-
Репликация данных
-
Консистентность
-
Разрешение рассинхронизации
-
Обработка сбоев
-
Архитектурная схема системы
-
Путь записи
-
Путь чтения
Материал ниже в большой степени основан на трех популярных key-value системах: Dynamo [4], Cassandra [5] и BigTable [6].
Разбиение данных
Для крупных приложений невозможно разместить весь набор данных на одном сервере. Самый простой способ — разбить данные на меньшие части и хранить их на нескольких серверах. При partitioning возникают две задачи:
-
Равномерно распределить данные по множеству серверов.
-
Минимизировать перемещение данных при добавлении или удалении узлов.
Консистентное хеширование, которое мы разбирали в предыдущей главе, отлично решает эти задачи. Кратко напомним его работу.
-
Сначала серверы размещаются на hash ring. На рисунке 4 восемь серверов, обозначенных s0, s1, …, s7, расположены на hash ring.
-
Затем ключ хешируется на то же кольцо и сохраняется на первом сервере, встреченном при движении по часовой стрелке. Например, по этой логике key0 хранится на s1.
Рисунок 4
Использование consistent hashing для partitioning дает следующие преимущества:
Automatic scaling: серверы можно автоматически добавлять и удалять в зависимости от нагрузки.
Heterogeneity: число virtual nodes для сервера пропорционально его мощности. Например, более производительные серверы получают больше virtual nodes.
Репликация данных
Чтобы обеспечить high availability и reliability, данные должны асинхронно реплицироваться на N серверов, где N — настраиваемый параметр. Эти N серверов выбираются так: после того как ключ отображен в точку на hash ring, мы идем от этой точки по часовой стрелке и выбираем первые N серверов для хранения копий данных. На рисунке 5 (N = 3) ключ key0 реплицируется на s1, s2 и s3.
Рисунок 5
При использовании virtual nodes первые N узлов на кольце могут принадлежать менее чем N физическим серверам. Чтобы избежать этой проблемы, при обходе по часовой стрелке выбираются только уникальные физические серверы.
Узлы в одном и том же дата-центре часто выходят из строя одновременно из-за отключения электричества, сетевых проблем, стихийных бедствий и т. д. Для большей надежности реплики размещают в разных дата-центрах, соединенных высокоскоростной сетью.
Консистентность
Поскольку данные реплицируются на нескольких узлах, их нужно синхронизировать между репликами. Quorum consensus позволяет обеспечить consistency как для чтений, так и для записей. Для начала введем несколько обозначений.
N = число реплик
W = write quorum размера W. Чтобы операция записи считалась успешной, нужно получить подтверждение как минимум от W реплик.
R = read quorum размера R. Чтобы операция чтения считалась успешной, нужно дождаться ответов как минимум от R реплик.
Рассмотрим пример на рисунке 6, где N = 3.
Рисунок 6 (ACK = acknowledgement)
W = 1 не означает, что данные записываются только на один сервер. Например, в конфигурации с
рисунка 6 данные реплицируются на s0, s1 и s2. W = 1 означает, что coordinator должен
получить хотя бы одно подтверждение, чтобы считать запись успешной. Например, если подтверждение
пришло от s1, дальше уже не нужно ждать ACK от s0 и s2. Coordinator выступает прокси
между клиентом и узлами.
Конфигурация W, R и N — это классический компромисс между задержкой и consistency. Если W = 1 или R = 1, операция завершается быстро, потому что coordinator ждет ответ только от одной реплики. Если W или R > 1, система обеспечивает лучшую consistency, но запрос будет медленнее, потому что coordinator придется ждать самого медленного участника.
Если W + R > N, strong consistency гарантируется, потому что обязательно существует хотя бы один пересекающийся узел, у которого есть самые свежие данные.
Как выбирать N, W и R под конкретные сценарии? Возможны, например, такие варианты:
-
Если R = 1 и W = N, система оптимизирована под быстрое чтение.
-
Если W = 1 и R = N, система оптимизирована под быструю запись.
-
Если W + R > N, strong consistency гарантируется (обычно N = 3, W = R = 2).
-
Если W + R <= N, strong consistency не гарантируется.
В зависимости от требований можно подбирать W, R, N так, чтобы получить нужный уровень consistency.
Модели консистентности
Модель consistency — еще один важный фактор при проектировании key-value store. Consistency model определяет степень согласованности данных. Возможен широкий спектр моделей:
-
Strong consistency: любое чтение возвращает значение, соответствующее результату самой свежей записи. Клиент никогда не видит устаревшие данные.
-
Weak consistency: последующие чтения могут не увидеть самое новое значение.
-
Eventual consistency: это частный случай weak consistency. Если дать системе достаточно времени, все обновления будут распространены, и все реплики придут к одному состоянию.
Strong consistency обычно достигается тем, что репликам запрещают принимать новые чтения и записи, пока все реплики не согласуют текущую запись. Такой подход плохо подходит для highly available систем, потому что может блокировать новые операции. Dynamo и Cassandra используют eventual consistency, и именно эту модель мы рекомендуем для нашего key-value store. При concurrent writes eventual consistency допускает появление inconsistent values в системе, а затем заставляет клиента прочитать несколько версий и разрешить конфликт. В следующем разделе разберем, как это делается через versioning.
Разрешение рассинхронизации: versioning
Репликация дает high availability, но вызывает inconsistency между репликами. Для решения этой проблемы используются versioning и vector clocks. Versioning означает, что каждое изменение данных рассматривается как новая неизменяемая версия. Перед тем как говорить о versioning, разберем пример возникновения inconsistency.
Как показано на рисунке 7, у обеих реплик n1 и n2 одно и то же значение. Назовем его исходным value. Сервер 1 и сервер 2 получают одинаковый результат для операции get("name").
Рисунок 7
Затем server 1 меняет имя на johnSanFrancisco, а server 2 — на johnNewYork, как показано
на рисунке 8. Эти два изменения происходят одновременно. Теперь у нас есть конфликтующие значения
— версии v1 и v2.
Рисунок 8
В этом примере исходное значение можно игнорировать, потому что обе модификации были основаны на нем. Но между двумя последними версиями нет очевидного способа выбрать победителя. Чтобы разрешить такой конфликт, нужна система версий, умеющая обнаруживать конфликт и согласовывать его. Для этого часто используют vector clock. Посмотрим, как он работает.
Vector clock — это пара [server, version], связанная с элементом данных. Она позволяет определить, предшествует ли одна версия другой, следует ли за ней или конфликтует с ней.
Предположим, vector clock записывается как D([S1, v1], [S2, v2], …, [Sn, vn]), где D — элемент данных, v1 — счетчик версии, а S1 — номер сервера. Если элемент D записывается на сервер Si, система должна сделать одно из двух:
-
Увеличить vi, если [Si, vi] уже существует.
-
Иначе создать новую запись [Si, 1].
Эта абстрактная логика поясняется конкретным примером на рисунке 9.
Рисунок 9
-
Клиент записывает элемент данных D1 в систему. Запись обрабатывает сервер Sx, и теперь у него есть vector clock D1([Sx, 1]).
-
Другой клиент читает последнее значение D1, обновляет его до D2 и записывает обратно. D2 является потомком D1, поэтому перезаписывает D1. Предположим, запись снова обрабатывает Sx, и теперь получается vector clock D2([Sx, 2]).
-
Еще один клиент читает D2, обновляет его до D3 и записывает обратно. Предположим, запись обрабатывает сервер Sy, и теперь vector clock равен D3([Sx, 2], [Sy, 1]).
-
Еще один клиент читает D2, обновляет его до D4 и записывает обратно. Предположим, запись обрабатывает сервер Sz, и теперь получается D4([Sx, 2], [Sz, 1]).
-
Когда следующий клиент читает D3 и D4, он обнаруживает конфликт, возникший из-за того, что элемент D2 был изменен и на Sy, и на Sz. Конфликт разрешается на стороне клиента, после чего обновленные данные отправляются на сервер. Предположим, запись обрабатывает Sx, и появляется D5([Sx, 3], [Sy, 1], [Sz, 1]). Чуть ниже мы поясним, как именно обнаруживается конфликт.
С помощью vector clock легко определить, что версия X является предком версии Y (то есть конфликта нет), если счетчики версий для каждого участника в vector clock версии Y больше либо равны соответствующим счетчикам в версии X. Например, D([s0, 1], [s1, 1]) — предок версии D([s0, 1], [s1, 2]). Поэтому конфликт не фиксируется.
Аналогично, версия X является sibling-версией Y (то есть конфликт существует), если хотя бы у одного участника в vector clock версии Y счетчик меньше, чем соответствующий счетчик в X. Например, эти два vector clock означают конфликт: D([s0, 1], [s1, 2]) и D([s0, 2], [s1, 1]).
Хотя vector clocks и позволяют разрешать конфликты, у них есть два заметных недостатка. Во-первых, они усложняют клиент, потому что клиенту приходится реализовывать логику conflict resolution.
Во-вторых, пары [server: version] в vector clock могут быстро разрастаться. Чтобы справиться с этой проблемой, задают порог длины, и если он превышен, самые старые пары удаляются. Это может снижать точность reconciliation, потому что отношение «предок-потомок» уже нельзя определить корректно. Однако, согласно статье про Dynamo [4], Amazon в production с этой проблемой не столкнулся, поэтому для большинства компаний это, вероятно, приемлемое решение.
Обработка сбоев
Как и в любой крупной системе под высокой нагрузкой, сбои не просто неизбежны — они обычны. Поэтому обработка failure scenarios крайне важна. В этом разделе сначала разберем техники обнаружения отказов, а затем посмотрим на типичные стратегии их обработки.
Обнаружение сбоев
В распределенной системе недостаточно считать сервер упавшим только потому, что другой сервер так сказал. Обычно требуется как минимум два независимых источника информации, чтобы признать сервер недоступным.
Как показано на рисунке 10, all-to-all multicasting — это прямолинейное решение. Но при большом количестве серверов оно неэффективно.
Рисунок 10
Лучшее решение — использовать децентрализованные методы обнаружения отказов, например gossip protocol. Gossip protocol работает так:
-
Каждый узел поддерживает membership list, содержащий member ID и heartbeat counters.
-
Каждый узел периодически увеличивает свой heartbeat counter.
-
Каждый узел периодически рассылает heartbeats случайному набору узлов, а те, в свою очередь, распространяют их дальше.
-
Когда узлы получают heartbeats, membership list обновляется до актуального состояния.
-
Если heartbeat не увеличивается дольше некоторого заранее заданного интервала, узел считается офлайн.
Рисунок 11
Как показано на рисунке 11:
-
Узел s0 хранит membership list, показанный слева.
-
Узел s0 замечает, что heartbeat counter узла s2 (
member ID = 2) долго не увеличивается. -
Узел s0 рассылает heartbeats, включающие информацию о s2, случайному набору узлов. Когда другие узлы подтверждают, что heartbeat counter s2 давно не обновлялся, узел s2 помечается как down, и эта информация распространяется дальше.
Обработка временных сбоев
После того как сбой обнаружен с помощью gossip protocol, система должна задействовать механизмы обеспечения availability. В strict quorum-подходе операции чтения и записи могут блокироваться, как мы уже видели в разделе о quorum consensus.
Чтобы повысить availability, используется техника “sloppy quorum” [4]. Вместо строгого соблюдения quorum система выбирает первые W здоровых серверов для записей и первые R здоровых серверов для чтений на hash ring. Недоступные серверы просто пропускаются.
Если сервер временно недоступен из-за сетевых проблем или отказа машины, запросы временно обрабатывает другой сервер. Когда исходный сервер снова поднимается, изменения передаются ему обратно для восстановления consistency. Этот процесс называется hinted handoff. Поскольку s2 недоступен на рисунке 12, чтения и записи временно обслуживает s3. Когда s2 возвращается в онлайн, s3 передает накопленные данные обратно s2.
Рисунок 12
Обработка постоянных сбоев
Hinted handoff помогает при временных сбоях. Но что делать, если реплика недоступна навсегда? Для этого используется anti-entropy protocol, поддерживающий синхронизацию реплик. Anti-entropy сравнивает данные на репликах и обновляет каждую из них до самой новой версии. Для обнаружения inconsistency и уменьшения объема передаваемых данных используется Merkle tree.
Согласно Wikipedia [7], «hash tree, или Merkle tree, — это дерево, в котором каждая внутренняя вершина помечена хешем меток или значений своих дочерних вершин. Hash trees позволяют эффективно и надежно проверять содержимое больших структур данных».
Предположим, пространство ключей — от 1 до 12. Ниже показано, как строится Merkle tree. Подсвеченные блоки обозначают inconsistency.
Шаг 1: разбить key space на buckets (в нашем примере 4), как показано на рисунке 13. Bucket используется как корневой узел уровня, чтобы ограничить глубину дерева.
Рисунок 13
Шаг 2: после создания buckets захешировать каждый ключ в bucket с помощью равномерной хеш-функции, как показано на рисунке 14.
Рисунок 14
Шаг 3: создать по одному hash node на bucket (рисунок 15).
Рисунок 15
Шаг 4: построить дерево вверх до корня, вычисляя хеши дочерних узлов, как на рисунке 16.
Рисунок 16
Чтобы сравнить два Merkle tree, сначала сравнивают корневые хеши. Если они совпадают, у серверов одинаковые данные. Если корневые хеши различаются, сравнивают левые дочерние хеши, затем правые и так далее. Так можно пройти по дереву и точно определить, какие buckets рассинхронизированы, а синхронизировать только их.
При использовании Merkle tree объем данных, которые нужно синхронизировать, пропорционален различиям между двумя репликами, а не общему объему данных. В реальных системах размер bucket довольно большой. Например, возможна конфигурация с одним миллионом buckets на один миллиард ключей, то есть каждый bucket содержит всего около 1000 ключей.
Обработка отказа дата-центра
Отказ дата-центра может произойти из-за отключения питания, сетевого сбоя, стихийного бедствия и т. д. Чтобы система переживала такие аварии, важно реплицировать данные между несколькими дата-центрами. Даже если один дата-центр полностью выходит из строя, пользователи все равно смогут доступаться до данных через другие дата-центры.
Архитектурная схема системы
Теперь, когда мы обсудили основные технические идеи для key-value store, можно перейти к общей архитектурной схеме, показанной на рисунке 17.
Рисунок 17
Основные свойства архитектуры:
-
Клиенты взаимодействуют с key-value store через простые API: get(key) и put(key, value).
-
Coordinator — это узел, который работает как прокси между клиентом и key-value store.
-
Узлы распределены по кольцу с использованием consistent hashing.
-
Система полностью децентрализована, поэтому добавление и перемещение узлов может происходить автоматически.
-
Данные реплицируются на нескольких узлах.
-
Единой точки отказа нет, поскольку у всех узлов одинаковый набор обязанностей.
Поскольку дизайн децентрализован, каждый узел выполняет множество задач, как показано на рисунке 18.
Рисунок 18
Путь записи
На рисунке 19 показано, что происходит после того, как запрос на запись попадает в конкретный узел. Обрати внимание: предложенные пути записи и чтения в значительной степени основаны на архитектуре Cassandra [8].
Рисунок 19
-
Запрос на запись сначала сохраняется в commit log file.
-
Затем данные попадают в memory cache.
-
Когда memory cache заполняется или достигает заданного порога, данные сбрасываются на диск в SSTable [9]. Напомним: sorted-string table (SSTable) — это отсортированный список пар
<key, value>.
Путь чтения
После того как запрос на чтение попадает в конкретный узел, система сначала проверяет, есть ли данные в memory cache. Если есть, данные сразу возвращаются клиенту, как показано на рисунке 20.
Рисунок 20
Если данных в памяти нет, они будут извлечены с диска. Для этого нужен эффективный способ понять, в какой именно SSTable находится нужный ключ. Для решения этой задачи обычно используется Bloom filter [10].
Путь чтения в случае отсутствия данных в памяти показан на рисунке 21.
Рисунок 21
-
Система сначала проверяет, есть ли данные в памяти. Если нет, переходит к шагу 2.
-
Если данных нет в памяти, система проверяет Bloom filter.
-
Bloom filter помогает определить, в каких SSTables может находиться ключ.
-
SSTables возвращают данные.
-
Результат возвращается клиенту.
Итоги
В этой главе разобрано много концепций и техник. Чтобы быстро освежить память, ниже приведена сводная таблица с задачами и соответствующими решениями для распределенного key-value store.
| Цель / проблема | Техника |
| Хранение больших объемов данных | Использовать consistent hashing для распределения нагрузки по серверам |
| Высокодоступные чтения | Репликация данных Размещение в нескольких дата-центрах |
| Высокодоступные записи | Versioning и conflict resolution с помощью vector clocks |
| Разбиение набора данных | Consistent Hashing |
| Постепенное масштабирование | Consistent Hashing |
| Heterogeneity | Consistent Hashing |
| Настраиваемая консистентность | Quorum consensus |
| Обработка временных сбоев | Sloppy quorum и hinted handoff |
| Обработка постоянных сбоев | Merkle tree |
| Обработка отказа дата-центра | Междатацентровая репликация |
Таблица 2
Справочные материалы
[1] Amazon DynamoDB: https://aws.amazon.com/dynamodb/
[2] memcached: https://memcached.org/
[3] Redis: https://redis.io/
[4] Dynamo: Amazon’s Highly Available Key-value Store:
https://www.allthingsdistributed.com/files/amazon-dynamo-sosp2007.pdf
[5] Cassandra: https://cassandra.apache.org/
[6] Bigtable: A Distributed Storage System for Structured Data:
https://static.googleusercontent.com/media/research.google.com/en//archive/bigtable-osdi06.pdf
[7] Merkle tree:
https://en.wikipedia.org/wiki/Merkle_tree
[8] Cassandra architecture:
https://cassandra.apache.org/doc/latest/architecture/
[9] SSTable:
https://www.igvita.com/2012/02/06/sstable-and-log-structured-storage-leveldb/
[10] Bloom filter:
https://en.wikipedia.org/wiki/Bloom_filter