Блог

Я разрабатываю хранилище событий на AWS и выбрал DynamoDB, потому что это показалось мне лучшим вариантом. Мой дизайн кажется неплохим, но я сталкиваюсь с некоторыми проблемами, которые не могу решить.

** Дизайн

События однозначно идентифицируются парой (StreamId, EventId) :

• StreamId : то же самое относится и к aggregateId, что означает один поток событий для одного агрегата.
• EventId : инкрементное число, которое помогает сохранить порядок внутри одного и того же потока событий.

События сохраняются в DynamoDB. Каждое событие сопоставляется с одной записью в таблице, где обязательными полями являются StreamID, EventID, eventName, Payload (можно легко добавить дополнительные поля).

Ключ разделения — это идентификатор потока, ключ сортировки — идентификатор события.

Оптимистичная блокировка используется при записи события в поток событий. Чтобы добиться этого, я использую условную запись DynamoDB. Если событие с тем же (StreamID, EventID) уже существует, мне нужно пересчитать агрегат, перепроверить бизнес-условия и, наконец, записать еще раз, если бизнес-условия пройдут.

Потоки событий

Каждый поток событий идентифицируется с помощью PartitionKey . Запрос потока для всех событий равен запросу PartitionKey=${StreamID} и ключу сортировки от 0 до MAX_INT.

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

Публикация событий

События публикуются с использованием комбинации потоков DynamoDB Lambda.

Повторное воспроизведение событий

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

Я думал об использовании дополнительной записи где-нибудь в DynamoDB, которая хранит в массиве все идентификаторы потоков. Затем я могу запросить его и начать запрашивать события, но если во время воспроизведения будет создан новый поток, я его потеряю.

Я что-то упустил? Или мой дизайн просто неправильный?

Вы можете использовать GSI для получения событий за заданный период времени. В зависимости от количества обрабатываемых событий вам может потребоваться записать сегмент GSI, чтобы избежать горячих клавиш. Предполагая, что элементы события составляют менее 1 КБ, вам нужно будет распределить их по GSI, если скорость приема превышает 1000 элементов в секунду. Если размер событий превышает 1 КБ, вам нужно будет распределить их больше. Для элементов размером менее 1 КБ возьмите общее количество событий в секунду и разделите на 1000. Это подскажет вам, сколько сегментов требуется GSI для работы с таблицей, например, если предположить, что вы проглатываете 5 тысяч событий в секунду, вам понадобится 5 сегментов.

При записи событий в таблицу добавьте новый атрибут с именем «GSIKey» и создайте случайное значение между 0-4 для этого атрибута при вставке событий. Создайте GSI, используя «GSIKey» в качестве ключа раздела и временную метку в качестве ключа сортировки. Когда вам нужно получить все события за заданный временной диапазон, запросите все 5 сегментов с нужным временным диапазоном, а затем просто объедините и отсортируйте результирующие наборы, чтобы получить упорядоченный по времени список событий. Если вы обрабатываете менее 1000 событий в секунду, вы можете использовать «0» в качестве значения GSIKey и просто запросить в этом разделе нужные вам события.

Я что-то упустил?

Не совсем; это сложная проблема [tm].

Ваши варианты использования при записи обычно касаются только одной ссылки в модели — указателя на текущую историю событий. Ваши варианты использования для чтения часто связаны с данными, распределенными по нескольким потокам.

Обычно это работает так: в вашем хранилище сохраняемых данных сохраняются не только записанные изменения, но и индекс, поддерживающий чтение. Например, хранилище сообщений postgres в Eventide зависит от индексации, которая происходит при вставке строк в таблицу. В случае хранилища событий обновления индекса записываются как часть той же сериализованной «транзакции», что и изменения в потоке (ах).

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

Уберите координацию, и у вас получится нечто аналогичное назначению уникального хоста для каждого потока.

Может быть полезно внимательно изучить базу данных объектов Git и ознакомиться с тем, что на самом деле происходит в этом хранилище под обложками. Я также обнаружил, что в докладе Рича Хикки «Язык системы» содержатся полезные концепции для отличия values от names от references .

Я выбрал DynamoDB, потому что это показалось мне лучшим вариантом

Если у вас нет веских деловых причин для создания своего хранилища событий с нуля, я бы посоветовал вам вместо этого взглянуть на Aurora и посмотреть, как далеко вы можете продвинуться с этим. Это может сэкономить время, необходимое для ожидания, пока кто-то другой создаст для вас экономичное облачное хранилище событий.

Оригинальный ответ: не могли бы вы подробнее рассказать о «агрегате»? Это то же самое, что EventID, или это отдельный атрибут элемента?

Вам нужно хранить события и агрегированные данные?

Каковы ваши требования к долговечности событий?

Если <14 дней, будет ли для вас вариант хранения событий в Kinesis? Как отметил Рик Хулихан, в вашем проекте могут возникнуть проблемы с горячими разделами или горячими ключами, что потребует от вас увеличения RCU / WCU в вашей таблице DynamoDB. Kinesis решает эту проблему. Его использование позволит вам сосредоточиться на логике вашего приложения.

Я буду рад помочь, если вы хотите, если вы могли бы поделиться более подробной информацией.

Обновление 4/23:

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

Вам нужно будет продумать эквивалентную логику условной записи, которую вы описали выше, для таблиц DynamoDB.

Преимущество CWL в том, что вы избежите проблем с горячими клавишами, о которых говорилось выше. Недостатки: (1) Вам нужно будет продумать решение против CWL. (2) DynamoDB предлагает задержку P99 <10 мс для чтения и <20 мс для записи. Время операций чтения и записи CWL намного выше (например, от 10 до 100 секунд или мс).

Я надеюсь, что это немного поможет.