Cassandra — ведра, кучи и снежки

Технологии не стоят на месте, а задачи перед ИТ-специалистами не меняются. А это означает, что настало время, когда каждый может делать крутые фичи наравне с крупными ИТ-компаниями. Компания ShuffleRow расскажет вам как сделать ленту новостей, в которую могут тысячами писать свои посты пользователи вашего сервиса.

Новая надежда

Что нам для этого может понадобиться? Рецепт простой: одна красивая девушка и ведра снега, которые сложены в кучи. И все это, конечно, нужно приправить своим любимым языком программирования.

Начнем с самого интересного: с того, кто она, эта красивая девушка? А имя ей Cassandra — распределенная, отказоустойчивая и рассчитанная на создание высоко масштабируемых и надежных хранилищ огромных массивов данных.

Уточним нашу задачу: условный пользователь пишет свои мысли, добавляет картинку, и остальные пользователи смотрят на это безобразие. Что сразу приходит в голову: даем пользователю идентификатор, сообщение — это просто текст, картинку сохраняем как ссылку на статику, запоминаем дату и время создания поста. По итогам получаем следующую структуру данных:

user_id int, message text, image_url text, created_at timestamp

Отлично, попробуем вложить это все в кассандру. Данные в кассандре хранятся как бесконечно расширяемая строка, в которой обязательно должен быть первичный ключ, и могут быть добавлены ключи раздела. При этом первичный ключ является указателем на конкретную строку с данными.

Логично предположить, что первичным ключом должен стать идентификатор пользователя user_id (primary key). Но, как всегда, есть нюансы: сортировка данных возможна только по ключам разделов. Предположим, что нюансы нам ясны: добавляем ключом раздела колонку (cluster key) created_at с сортировкой в обратном порядке. В итоге имеем следующую структуру таблицы:

схематичное представление данных в полученной таблице

Попробуем теперь разобраться с тем, как данные на самом деле хранятся в таблице в кассандре. Предположим, что два пользователя пишут посты в нашу ленту, первый пользователь написал два поста, а второй три поста. Посты каждого пользователя сохраняются в строки, которые соответствуют строкам с их идентификаторами. Каждый новый пост дописывается в начало соответствующей строки. При этом ключ колонок является композицией ключа раздела и названия колонки.

слева — реляционное представление данных в таблице, справа — схема хранения данных в Cassandra

Кассандра наносит ответный удар

Отлично, благодаря двум нашим пользователям у нас появились данные в нашей таблице. Попробуем их отобразить в ленте постов, чтобы остальные пользователи могли прочесть их и, возможно, поделиться своими новостями. Для этого выполним запрос select * from feed

Что мы получили? Реальная выдача сильно отличается от того, что мы ожидали, если бы работали с реляционной базой данных, такой как PostgresQL или MySQL. И опять нюансы: так как первичным ключом нашей таблицы является поле user_id, кассандра сначала достает все строки, а уже потом берет все данные из каждой строки, которые отсортированы по ключу раздела created_at.

Итак, выбранная нами архитектура таблицы для ленты постов не подходит. Первая мысль, которая приходит в голову — это сделать первичным ключом дату и время создания поста, а user_id ключом раздела. Но в таком случае мы получим большое количество строк, в которых будет по одной записи, что, в свою очередь, сведет на нет все преимущества использования кассандры. То есть необходимо каким-то образом сгруппировать посты наших пользователей в кучи, при этом сохранив сортировку по дате и времени.

Если бы мы использовали классическую базу данных, то первая колонка, которую бы мы добавили в таблицу, была бы идентификатором самого поста. Стандартный, авто-инкрементальный целочисленный идентификатор. Не трудно догадаться, что, имея такой идентификатор, мы могли бы решить проблему сортировки постов. Так как кассандра является распределенной базой данных, она не предоставляет никаких механизмов генерации целочисленных авто-инкрементальных идентификаторов. Следовательно, клиент сам должен генерировать идентификатор записи и заботится о возможных коллизиях.

Рассмотрим варианты генерации идентификатора записи:

• использовать в качестве генератора классическую базу данных, такую как PostgresQL или MySQL,
• использовать механизм генерации идентификаторов MongoDB,
• использовать механизм счетчиков Redis или ему подобный,
• генерировать идентификатор, основываясь на текущей дате и времени;

Самым адекватным выглядит последний вариант в силу того, что время имеет свойство только увеличиваться в физическом его понимании, и, следовательно, каждый новый пост наших пользователей будет всегда иметь дату и время создания, которое будет либо больше, чем у предыдущих постов, либо равно ему. Если выбрать вариант генерации идентификатора на основе времени, мы также не будем зависеть от дополнительных технических систем и не будем делать лишние вызовы в момент создания поста. Итак, для того, чтобы взять в работу генерацию идентификатора на основе даты и времени создания поста, остается разобраться с последним нюансом: что делать, когда несколько пользователей создают свои посты в один день и в одно время, с точностью до миллисекунд?

К счастью данную проблему решила известная IT-компания Twitter. Для данных целей был разработан алгоритм, который называется Snowflake. Это 64 битный номер, в котором:

• 1 бит (sign) — необходим для определения границ timestamp,
• 41 бит (timestamp) — дата генерации id в микросекундах,
• 10 бит (generator) — id сервиса генерирующего id. Обычно разделяют на 2 — Datacenter ID (5 бит) и Machine ID (5 бит),
• 12 бит (sequence) — инкрементируемое число.

Инкрементируется в тех случаях, когда timestamp генерируемого id, совпадает с timestamp последнего сгенерированного id. Своего рода защита от коллизии на локальном уровне.

Отлично, у нас появилась возможность генерировать идентификаторы для постов в целочисленном виде с инкрементацией значений, основываясь на времени. А это подводит нас к выводу, что мы можем группировать все посты всех пользователей по временному признаку. И тут приходят они — бакеты. Предположим, что все посты наших пользователей равномерно распределены во времени. Это означает, что в качестве бакета может выступать дата создания поста, а генерировать его мы можем на основании идентификатора поста.

Итак, что мы имеем? У нас есть идентификатор поста, некое агрегирующее их во времени значение, идентификатор пользователя и данные содержания поста. Опишем структуру модернизированной таблицы для постов:

схематичное представление данных в модернизированной таблицы feed

Предположим, что предыдущие два пользователя написали еще постов и мы снова наполнили свою таблицу данными. Запросим из нашей новой таблички данные и посмотрим, что кассандра нам выдаст в этот раз. Для этого выполним запрос select * from feed еще раз.

Очевидно, что, хотя архитектурно мы добились значительных успехов, это не позволило нам достичь правильных результатов выборки. Логичным будет предположить, что достаточно добавить в запрос конструкцию order by. Попробуем это сделать и выполним запрос

select * from feed order by bucket desc, snowflake desc

Результатом будет ошибка:

ORDER BY is only supported when the partition key is restricted by an EQ or an IN

Кассандра говорит нам, что сортировка возможна только при условии, что запрос будет отфильтрован по первичному ключу, иными словами, по бакету. Но встает вопрос: где взять значение нужного бакета для запроса? Потому что как бы данные не хранились, важно вывести список последних постов пользователей в порядке от самого нового к самому старому.

Возвращение сортировки

Попробуем прикинуть возможные алгоритмы того, как нам получить последний бакет, сохраненный в кассандре:

1. сгенерировать бакет на основе текущей даты и выполнить запрос вида select * from feed where bucket = <current bucket>. Если выдача будет пустая, или строк в выдаче будет недостаточно для выборки, то далее необходимо сгенерировать новый бакет на основе даты предыдущей текущей и повторить данный алгоритм;
2. при записи нового поста в таблицу feed сохранять бакет из этой записи в другую таблицу. Далее перед запросом данных из таблицы feed запрашивать последний бакет из вспомогательной таблицы, после чего выполнять запрос вида select * from feed where bucket = <current bucket>. Если выдача будет пустая, или строк в выдаче будет недостаточно для выборки, то далее необходимо достать из вспомогательной таблицы следующий бакет и повторить данный алгоритм.

Для решения нашей задачи могут подойти оба варианта, причем первый вариант даже предпочтительней, хотя и выглядит как костыль из двухтысячных. Второй вариант был бы хорошим, если мы имели бы ограничения по выводу постов. Например, таким ограничением может являться выдача постов для сообщества по типу социальной сети.

Попробуем выполнить запрос в кассандру для получения трех последних постов, которые написали наши пользователи. Предположим, что мы преобразовали текущую дату в бакет и получили значение 3. Запрос выглядит следующим образом:

select * from feed where bucket = 3 limit 3

Очевидно, что в нашем примере за текущий день (бакет 3) есть всего два поста, следовательно, нам для получения полной выборки данных необходимо посмотреть посты за день предыдущий, т.е. выполнить такой же запрос, но для бакета 2. Но за текущий день мы получили две записи и нам необходимо добрать лишь одну запись из предыдущего дня. Для этого выполняем запрос:

select * from feed where bucket = 2 limit 1

Используя наш любимый язык программирования мы с легкостью можем склеить два массива в один и получить полную выборку последних трех постов наших пользователей. Но всплывает еще один нюанс. Что делать, если понадобится достать следующие три поста? Иными словами, мы с вами имеем три бакета, данные в которых распределены почти равномерно, но их кол-во нам не известно. Т.е. в первом бакете мы имеем одну запись, во втором три, а в третьем две. Если последний, третий бакет, мы полностью получили, то во втором у нас еще остались две записи, и если следовать логике показа постов по три штуки, то для того, чтобы подгрузить оставшихся три, нам необходимо достать из таблицы два последних поста из второго бакета и один пост из первого бакета.

В рамках работы кассандры это задача нетривиальная. Но так как мы используем snowflake для генерации идентификаторов постов, нам достаточно модернизировать наш запрос ограничением по идентификатору последнего поста, который мы достали в предыдущей пачке. Наш запрос для подгрузки оставшихся постов для второго бакета будет выглядеть следующим образом:

select * from feed where bucket = 2 and snowflake < 4 limit 3

а второй запрос для получения полной выборки

select * from feed where bucket = 1 and snowflake < 4 limit 1

Подводя итоги

Что мы получили в итоге? Рабочую архитектуру хранения и выдачи постов для ленты. Работая с кассандрой мы можем быстро сохранять посты наших пользователей в больших объемах, не теряя при этом в скорости их выдачи. Подобное решение мы давно и успешно используем в своем сервисе и надеемся, что данная статья поможем и вам внедрить успешный опыт в свои текущие и будущие разработки.