Apache Flink - доступ к внутреннему буферу WindowedStream из MapFunction другого потока

Соедините два потока, затем используйте CoProcessFunction. Вызов метода, который получает поток функций, может применить их к тому, что находится в окне вызова другого метода.

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

Предполагая, что Fx агрегирует входящие foo на лету, а Gx обрабатывает количество foo для окна, вы должны быть в состоянии достичь того, что хотите, следующим образом:

DataStream<Function> userRequest  = .... // request function from user
Iterator<Function> iter = DataStreamUtils.collect(userRequest);
Function Gx = iter.next();

DataStream<Foo> in  = .... // source data
 .keyBy(new MyKeySelector())
 .countWindow(100, 1)
 .fold(new ArrayList<>(), new MyFoldFunc(), new MyProcessorFunc(Gx))
 .addSink(new MySinkFunction())

Функция складывания (работает с входящими данными, как только они поступают) может быть определена следующим образом:

private static class MyFoldFunc implements FoldFunction<foo, Tuple2<Integer, List<foo>>> {
    @Override
    public Tuple2<Integer, List<foo>> fold(Tuple2<Integer, List<foo>> acc, foo f) {
        acc.f0 = acc.f0 + 1; // if Fx is a simple aggregation (count)
        acc.f1.add(foo);
        return acc;
    }
}

Функция процессора может быть примерно такой:

public class MyProcessorFunc
    extends ProcessWindowFunction<Tuple2<Integer, List<foo>>, Tuple2<Integer, FunctionResult>, String, TimeWindow> {

    public MyProcessorFunc(Function Gx) {
        super();
        this.Gx = Gx;
    }

    @Override
    public void process(String key, Context context,
                        Iterable<Tuple2<Integer, List<foo>> accIt,
                        Collector<Tuple2<Integer, FunctionResult>> out) {
        Tuple2<Integer, List<foo> acc = accIt.iterator().next();
        out.collect(new Tuple2<Integer, FunctionResult>(
            acc.f0, // your Fx aggregation
            Gx.apply(acc.f1), // your Gx results
        ));
    }
}

Обратите внимание, что функции fold \ reduce по умолчанию не буферизуют элементы внутри. Здесь мы используем свертку для вычисления метрик на лету, а также для создания списка элементов окна.

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

Примечание: вам может потребоваться добавить следующую зависимость для использования DataStreamUtils:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-contrib</artifactId>
    <version>0.10.2</version>
</dependency>