Разъяснение для государственного Backend

Question

Я читал документы Flink, и мне нужно было немного разъяснений. Надеюсь, кто-нибудь может помочь мне здесь.

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

The RocksDBStateBackend holds in-flight data in a RocksDB database that is (per default) stored in the TaskManager data directories

Относятся ли данные в полете к входящим данным, скажем, из потока kafka, который еще не был проверен?

Upon checkpointing, the whole RocksDB database will be checkpointed into the configured file system and directory. Minimal metadata is stored in the JobManager’s memory

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

Note that the amount of state that you can keep is only limited by the amount of disk space available

Значит ли это, что я мог бы выполнить аналитический запрос для потенциально высокого потока по всему потоку с очень ограниченными ресурсами. Предположим, что мой Kafka Stream имеет скорость 50 тыс. Сообщений в секунду, тогда я могу запустить его на одном ядре в своем кластере EMR, и компромисс будет в том, что Flink не сможет догнать входящую скорость и будет иметь отставание, но если на диске достаточно места, то не будет ошибки OOM?

Когда контрольная точка завершена, я предполагаю, что заполненные агрегированные метаданные контрольной точки (например, путь HDFS или S3 от каждого ТМ) со всех ТМ будут отправлены в JM ?. В случае сбоя TM JM раскрутит новый JM и восстановит состояние с последней контрольной точки.

The default setting for JM in flink-conf.yaml - jobmanager.heap.size: 1024m.
Моя путаница в том, почему JM требуется 1 Гб памяти в куче. Что все делает JM, кроме синхронизации между ТМ. Как я на самом деле решаю, сколько памяти должно быть сконфигурировано для JM на производстве.

Может ли кто-нибудь подтвердить, что мое понимание верно или нет, и указать мне правильное направление. Заранее спасибо!

Answer 1

В целом ваше понимание кажется правильным. Одно замечание: в случае сбоя TM JM раскручивает новый TM и восстанавливает состояние с последней контрольной точки (а не раскручивает новый JM ).

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

Менеджер заданий также выполняет и другие роли - это координатор контрольных точек и конечная точка API для веб-интерфейса и показателей.

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