Блог

Рассмотрим следующий пример класса, экземпляры которого хранятся в ListState :

 class BusinessObject {
  long clientId;
  String region;
  Instant lastDealDate; 
  bool isActive;
}
  

Приложение требует, чтобы этот объект не находился в состоянии flink, если прошло 1 год с момента заключения последней сделки ( lastDealDate ) с конкретным клиентом, и клиент не активен, т.е. isActive == false

Каков был бы правильный способ сделать это и сообщить flink об этих 2 факторах, чтобы он автоматически удалял эти записи? В настоящее время я считываю все элементы в состоянии, очищаю состояние, а затем добавляю обратно соответствующие, однако это начнет занимать много времени по мере увеличения числа клиентов и увеличения размера состояния. Большинство моих поисковых запросов в Интернете говорят об использовании time-to-live и настройке его для моего состояния через descriptor . Однако моя логика не может полагаться на время обработки / события / приема, и мне нужно также проверить, является ли isActive значение false.

Дополнительная информация: контекст не задан, а серверная часть — RocksDB. Причина ListState , по которой используется a, заключается в том, что все относительное состояние / история в соответствии с вышеуказанными условиями должны сбрасываться каждый день.

Есть предложения?

С помощью серверной части состояния RocksDB Flink может добавлять к ListState без прохождения сериализации / десериализации, но любое чтение или модификация, отличные от добавления, являются дорогостоящими из-за ser / de.

Вам будет лучше, если вы сможете переделать вещи так, чтобы эти BusinessObjects хранились в mapState, даже если вам иногда приходится перебирать всю карту. Каждая пара ключ / значение в mapState будет отдельной записью RocksDB, и вы сможете индивидуально создавать / обновлять / удалять их без необходимости проходить ser / de для всей карты (если вам не нужно ее сканировать). (Для чего это стоит, итерация по mapState в RocksDB выполняется по карте в порядке сортировки по сериализованному ключу.)

mapState доступно только как состояние с ключом (или широковещательное), поэтому для этого изменения потребуется ввести ключ потока. Использование keyBy вызывает перетасовку сети (и ser / de), поэтому это будет дорого, но не так дорого, как использование ListState.