фильтр по максимуму в поле кортежа в Apache Flink

Question

Я использую Apache Flink Streaming API для обработки файла данных, и мне интересно получить только результаты из последнего из windows. Есть ли способ сделать это? Если это невозможно, я подумал, что могу отфильтровать максимум первого поля в результирующем кортеже (значение Long).

SingleOutputStreamOperator<Tuple12<Long, String, String, Integer, String, Integer, String, Integer, String, Integer, String, Integer>> top5SlidingEventTimeWindowsFiltered = top5SlidingEventTimeWindows.filter(new FilterFunction<Tuple12<Long,String,String,Integer,String,Integer,String,Integer,String,Integer,String,Integer>>() {

        public boolean filter(
                Tuple12<Long, String, String, Integer, String, Integer, String, Integer, String, Integer, String, Integer> value)
                throws Exception {
            …

        }
    });

В приведенном выше преобразовании фильтрации было бы фильтровать по максимуму значение первого поля кортежа. Можно ли это как-то сделать?

Answer 1

В API DataStream, когда вы потребляете данные из конечного источника (например, файла), когда источник достигает конца своего ввода, он отправляет водяной знак со значением MAX_WATERMARK. Вы можете использовать это, чтобы определить, что задание выполнено.

Итак, в таком случае, как ваш, вы можете поставить ProcessFunction после windows, и он будет постоянно сохранять в состоянии последние полученные результаты. получил пока. Установите таймер для MAX_WATERMARK, и когда он появится, используйте то, что находится в состоянии, чтобы получить желаемый результат.

Это должно быть KeyedProcessFunction, иначе вы не сможете использовать таймеры. Если поток не имеет ключей, вам все равно придется ввести ключ - вы можете просто ввести константу, при условии, что вы не против иметь параллелизм, равный единице.