Flink窗口函数getResult未触发

Question

我正在尝试在我的 Flink 作业中使用事件时间，并使用 BoundedOutOfOrdernessTimestampExtractor 来提取时间戳并生成水印。 但是我有一些输入 Kafka 有稀疏流，它可以很长时间没有数据，这使得 AggregateFunction 中的 getResult 根本没有被调用。我可以看到数据进入add 函数。

我已设置getEnv().getConfig().setAutoWatermarkInterval(1000L); 我试过了

 eventsWithKey
            .keyBy(entry -> (String) entry.get(key))
            .window(TumblingEventTimeWindows.of(Time.minutes(windowInMinutes)))
            .allowedLateness(WINDOW_LATENESS)
            .aggregate(new CountTask(basicMetricTags, windowInMinutes))

还有会话窗口

eventsWithKey
            .keyBy(entry -> (String) entry.get(key))
            .window(EventTimeSessionWindows.withGap(Time.seconds(30)))
            .aggregate(new CountTask(basicMetricTags, windowInMinutes))

所有水印指标显示No Watermark 如何让 Flink 忽略无水印的事情？

Answer 1

仅供参考，这通常被称为“空闲源”问题。这是因为每当 Flink 算子有两个或多个输入时，它的水印是其输入中水印的最小值。如果其中一个输入停止，其水印将不再前进。

请注意，Flink 没有每个键的水印——一个给定的操作符通常跨多个键的事件多路复用。只要某些事件流经给定任务的输入流，它的水印就会前进，空闲键的事件时间计时器仍然会触发。要发生这种“空闲源”问题，任务必须有一个完全空闲的输入流。

如果您可以安排，最好的解决方案是让您的数据源包含 keepalive 事件。这将使您可以放心地推进您的水印，因为您知道源只是空闲而不是离线。

如果这是不可能的，并且如果您有一些资源不是空闲的，那么您可以在 BoundedOutOfOrdernessTimestampExtractor 前面（以及 keyBy 之前）放置一个 rebalance()，以便每个实例继续接收一些事件，并可以推进其水印。这是以额外的网络洗牌为代价的。

也许最常用的解决方案是使用水印生成器，它检测空闲并根据处理时间计时器人为地推进水印。 ProcessingTimeTrailingBoundedOutOfOrdernessTimestampExtractor 就是一个例子。

Answer 2

引入了具有空闲功能的新水印。 Flink 将在计算最小值时忽略这些空闲水印，因此将考虑包含数据的单个分区。 https://ci.apache.org/projects/flink/flink-docs-release-1.11/api/java/org/apache/flink/api/common/eventtime/WatermarksWithIdleness.html

Answer 3

我也有同样的问题 - 一个 src 可能会长时间处于非活动状态。
以下解决方案基于WatermarksWithIdleness。

这是一个展示该概念的独立 Flink 作业。

package com.demo.playground.flink.sleepysrc;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.eventtime.WatermarksWithIdleness;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;


public class SleepyJob {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final EventGenerator eventGenerator = new EventGenerator();
        WatermarkStrategy<Event> strategy = WatermarkStrategy.
                <Event>forBoundedOutOfOrderness(Duration.ofSeconds(5)).
                withIdleness(Duration.ofSeconds(Constants.IDLE_TIME_SEC)).
                withTimestampAssigner((event, timestamp) -> event.timestamp);
        final DataStream<Event> events = env.addSource(eventGenerator).assignTimestampsAndWatermarks(strategy);
        KeyedStream<Event, String> eventStringKeyedStream = events.keyBy((Event event) -> event.id);
        WindowedStream<Event, String, TimeWindow> windowedStream = eventStringKeyedStream.window(EventTimeSessionWindows.withGap(Time.milliseconds(Constants.SESSION_WINDOW_GAP)));
        windowedStream.allowedLateness(Time.milliseconds(1000));
        SingleOutputStreamOperator<Object> result = windowedStream.process(new ProcessWindowFunction<Event, Object, String, TimeWindow>() {
            @Override
            public void process(String s, Context context, Iterable<Event> events, Collector<Object> collector) {
                int counter = 0;
                for (Event e : events) {
                    Utils.print(++counter + ") inside process: " + e);
                }
                Utils.print("--- Process Done ----");
            }
        });
        result.print();
        env.execute("Sleepy flink src demo");
    }


    private static class Event {
        public Event(String id) {
            this.timestamp = System.currentTimeMillis();
            this.eventData = "not_important_" + this.timestamp;
            this.id = id;
        }

        @Override
        public String toString() {
            return "Event{" +
                    "id=" + id +
                    ", timestamp=" + timestamp +
                    ", eventData='" + eventData + '\'' +
                    '}';
        }

        public String id;
        public long timestamp;
        public String eventData;
    }

    private static class EventGenerator implements SourceFunction<Event> {

        @Override
        public void run(SourceContext<Event> ctx) throws Exception {
            /**
             *  Here is the sleepy src - after NUM_OF_EVENTS events are collected , the code goes to a SHORT_SLEEP_TIME sleep
             *  We would like to detect this inactivity and FIRE the window
             */
            int counter = 0;
            while (running) {
                String id = Long.toString(System.currentTimeMillis());
                Utils.print(String.format("Generating %d events with id %s", 2 * Constants.NUM_OF_EVENTS, id));
                while (counter < Constants.NUM_OF_EVENTS) {
                    Event event = new Event(id);
                    ctx.collect(event);
                    counter++;
                    Thread.sleep(Constants.VERY_SHORT_SLEEP_TIME);
                }
                // here we create a delay:
                // a time of inactivity where
                // we would like to FIRE the window
                Thread.sleep(Constants.SHORT_SLEEP_TIME);
                counter = 0;
                while (counter < Constants.NUM_OF_EVENTS) {
                    Event event = new Event(id);
                    ctx.collect(event);
                    counter++;
                    Thread.sleep(Constants.VERY_SHORT_SLEEP_TIME);
                }
                Thread.sleep(Constants.LONG_SLEEP_TIME);
            }
        }

        @Override
        public void cancel() {
            this.running = false;
        }

        private volatile boolean running = true;

    }

    private static final class Constants {
        public static final int VERY_SHORT_SLEEP_TIME = 300;
        public static final int SHORT_SLEEP_TIME = 8000;
        public static final int IDLE_TIME_SEC = 5;
        public static final int LONG_SLEEP_TIME = SHORT_SLEEP_TIME * 5;
        public static final long SESSION_WINDOW_GAP = 60 * 1000;
        public static final int NUM_OF_EVENTS = 4;
    }

    private static final class Utils {
        public static void print(Object obj) {
            System.out.println(new java.util.Date() + " > " + obj);
        }
    }
}

Answer 4

对于其他人，如果您使用的是 Kafka，请确保有数据来自您所有主题的分区

我知道这听起来很愚蠢，但就我而言，我只有一个来源，问题仍然存在，因为我在一个有 10 个分区的 Kafka 主题（单一来源）中使用非常少的数据进行测试。数据集太小了，以至于一些主题的分区没有什么可以提供的，虽然我只有一个来源（一个主题），但 Flink 并没有增加 Watermark。

当我将源切换到具有单个分区的主题时，水印开始前进。