Процессор Kafka Stateful Stream с хранилищем состояний: за кулисами

Question

Я пытаюсь понять Stateful Stream processor.

Насколько я понимаю, в этом типе потокового процессора он поддерживает какое-то состояние, используя State Store.

Я узнал, что одним из способов реализации State Store является использование RocksDB. Предполагая следующее topology (и только один процессор - stateful)

A-> B -> C; процессор B с сохранением состояния с локальным хранилищем состояний и changelog включен . Я использую API низкого уровня.

Предполагая, что sp прослушивает один kafka topi c, скажем, topic-1 с 10 разделами.

Я заметил, что при запуске приложения ( 2 экземпляра на разных физических машинах и num.stream.threads = 5), затем для state store создается структура каталогов, которая выглядит примерно так:

0_0, 0_1, 0_2 .... 0_9 (на каждой машине по пять итого 10 разделов).

Я просматривал некоторые онлайн-материалы, в которых говорилось, что мы должны создать StoreBuilder и присоединить его топологию, используя addStateStore() вместо из создания хранилища состояний внутри процессора.

Примерно:

topology.addStateStore(storeBuilder,"processorName")

Ref also: org.apache.kafka.streams.state.Store

Я не понял, в чем разница в присоединении storeBuilder к топологии против фактического создания хранилища состояний в процессоре . В чем разница между ними?

Вторая часть: для statestore он создает каталог вроде: 0_0, 0_1 et c. Кто и как его создает? Есть ли какое-то сопоставление 1: 1 между темами kafka (которые прослушивает sp) и количеством каталогов, которые создаются для State Store?

Answer 1

Я не понял, в чем разница между подключением storeBuilder к топологии и фактическим созданием хранилища состояний внутри процессора. В чем разница между ними?

Чтобы позволить Kafka Streams управлять хранилищем за вас (отказоустойчивость, миграция), Kafka Streams должен знать о хранилище. Таким образом, вы даете Kafka Streams StoreBuilder, а Kafka Streams создает и управляет хранилищем для вас.

Если вы просто создаете хранилище внутри своего процессора, Kafka Streams не знает о хранилище, и хранилище выиграет ' t быть отказоустойчивым.

Для statestore он создает каталог, например: 0_0, 0_1 et c. Кто и как его создает? Есть ли какое-то сопоставление 1: 1 между темами kafka (которые прослушивает sp) и количеством каталогов, создаваемых для State Store?

Да, сопоставление есть. Хранилище разделяется базой в количестве вводимых топи c разделов. Вы также получаете «задачу» для каждого раздела, а каталоги задач имеют имя y_z, где y является номером субтопологии, а z - номером раздела. Для вашей простой топологии у вас есть только одна под-топология, чтобы все каталоги, которые вы видите, имеют одинаковый префикс 0_.

Следовательно, ваше логическое хранилище имеет 10 физических сегментов. Это сегментирование позволяет Kafka Streams зеркально отображать состояние, когда соответствующий входной раздел topi c назначен другому экземпляру. В целом вы можете запустить до 10 экземпляров, каждый из которых будет обрабатывать один раздел и размещать один осколок вашего хранилища.