У меня есть пользовательский RichCoFlatMapFunction, который использует ValueState член. Документы говорят, что
интерфейс ключ / значение привязан к ключу текущего элемента ввода
Что, если я нажму на подключенные потоки следующим образом:
val connected = streamA
.connect(streamB)
.keyBy(a=>a.foo, b=>b.bar)
.flatMap(new MyRichCoFlatMapFunction)
Как тогда семантика? Это связано с первым, вторым или их комбинацией?
Первый аргумент keyBy (a => a.foo) определяет ключ первого потока (streamA). Второй аргумент (b => b.bar) ключ второго потока (streamB). Оба аргумента должны возвращать ключ одного и того же типа, т.е. типы a.foo и b.bar должны быть одинаковыми.
Затем оператор соединения отправит все записи из streamA и streamB с одинаковым ключом одному и тому же оператору. Состояние RichCoFlatMapFunction установит ValueState для ключа текущего элемента, то есть, если flatMap1(a: TypeA, out: Collector[TypeOut]) вызывается для значения из streamA, состояние устанавливается для ключа a.foo, а если flatMap2(b: TypeB, out: Collector[TypeOut]) вызывается для значения из streamB, состояние устанавливается для ключа b.bar.
