Как распараллелить работу Flink с Guava кешем?

Question

Я написал работу Flink, которая использует кеш Guava. Объект кэша создается и используется в функции run (), вызываемой в функции main ().

Это что-то вроде:

main() {
   run(some,params)
}

run() {
   //create and use Guava cache object here
}

Если я выполню это задание Flink с некоторым уровнем параллелизма, будут ли все параллельные задачи использовать один и тот же объект кэша? Если нет, то как я могу заставить их всех использовать один кеш?

Кэш используется внутри функции process () для потока. Так что это похоже на

incoming_stream.process(new ProcessFunction() { //Use Guava Cache here })  

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

Answer 1

Использование кэша Guava с Flink обычно является анти-паттерном. Не то чтобы его нельзя было заставить работать, но, возможно, есть более простое и масштабируемое решение.

Стандартный подход к дедупликации в полностью масштабируемом и быстродействующем режиме с помощью Flink состоит в том, чтобы разделить поток по некоторому ключу (используя keyBy), а затем использовать состояние ключа для запоминания увиденных ключей. Управляемое состояние Flink управляется Flink таким образом, что делает его отказоустойчивым и перекалиброванным, сохраняя его локальным. Ключевое состояние Flink - это хранилище ключей / значений, где каждый экземпляр обрабатывает все события для некоторой части пространства ключей. Вам гарантировано, что для каждого ключа все события для одного и того же ключа будут обрабатываться одним и тем же экземпляром, поэтому это хорошо работает для дедупликации.

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

Answer 2

Задачи Flink выполняются на нескольких JVM или машинах, поэтому проблема заключается в том, как обмениваться объектами между JVM.

Обычно вы можете получать объекты из удаленной JVM с помощью RPC (через tcp) или остальных (через http) вызовов.

В качестве альтернативы, вы можете сериализовать объекты и сохранить их в базе данных, как reids, затем прочитать из базы данных и десериализовать в объекты.

В Flink есть более изящный способ добиться этого, вы можете хранить объекты в состоянии , и broadcast_state может вам подойти.

Широковещательное состояние было введено для поддержки случаев использования, когда некоторые данные, поступающие из одного потока, должны быть переданы на все последующие задачи

Надеюсь, это поможет.