如何实现在 X 分钟内未收到任何事件后发出的 Flink 事件时间触发器

Question

我有点难以理解 Flink 触发器的工作原理。我的数据流包含带有我基于该 sessionId 聚合的 sessionId 的事件。每个会话将包含一个 Started 和 Ended 事件，但有时 Ended 事件会丢失。

为了处理这个问题，我设置了一个触发器，它会在处理结束事件时发出聚合会话。但是，如果 2 分钟内没有事件从该会话到达，我想发出到目前为止我们聚合的任何内容（我们发送事件的应用程序每分钟发送一次心跳，所以如果我们没有收到任何事件，则认为会话丢失） .

我设置了如下触发函数：

public class EventTimeProcessingTimeTrigger extends Trigger<HashMap, TimeWindow> {
    private final long sessionTimeout;
    private long lastSetTimer;

    // Max session length set to 1 day
    public static final long MAX_SESSION_LENGTH = 1000l * 86400l;

    // End session events
    private static ImmutableSet<String> endSession = ImmutableSet.<String>builder()
            .add("Playback.Aborted")
            .add("Playback.Completed")
            .add("Playback.Error")
            .add("Playback.StartAirplay")
            .add("Playback.StartCasting")
            .build();

    public EventTimeProcessingTimeTrigger(long sessionTimeout) {
        this.sessionTimeout = sessionTimeout;
    }

    @Override
    public TriggerResult onElement(HashMap element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        lastSetTimer = ctx.getCurrentProcessingTime() + sessionTimeout;
        ctx.registerProcessingTimeTimer(lastSetTimer);

        if(endSession.contains(element.get(Field.EVENT_TYPE))) {
            return TriggerResult.FIRE_AND_PURGE;
        }

        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
        return TriggerResult.FIRE_AND_PURGE;
    }

    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
        return time == window.maxTimestamp() ?
                TriggerResult.FIRE_AND_PURGE :
                TriggerResult.CONTINUE;
    }

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        ctx.deleteProcessingTimeTimer(lastSetTimer);
    }

    @Override
    public boolean canMerge() {
        return true;
    }

    @Override
    public void onMerge(TimeWindow window,
                        OnMergeContext ctx) {
        ctx.registerProcessingTimeTimer(ctx.getCurrentProcessingTime() + sessionTimeout);
    }
}

为了给事件设置水印，我使用应用程序设置的水印，因为 appEventTime 可能与服务器上的 wallClock 不同。我像这样提取水印：

DataStream<HashMap> playerEvents = env
                .addSource(kafkaConsumerEvents, "playerEvents(Kafka)")
                .name("Read player events from Kafka")
                .uid("Read player events from Kafka")
                .map(json -> DECODER.decode(json, TypeToken.of(HashMap.class))).returns(HashMap.class)
                .name("Map Json to HashMap")
                .uid("Map Json to HashMap")
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<HashMap>(org.apache.flink.streaming.api.windowing.time.Time.seconds(30))
                {
                    @Override
                    public long extractTimestamp(HashMap element)
                    {
                        long timestamp = 0L;
                        Object timestampAsObject = (Object) element.get("CanonicalTime");
                        timestamp = (long)timestampAsObject;
                        return timestamp;
                    }
                })
                .name("Add CanonicalTime as timestamp")
                .uid("Add CanonicalTime as timestamp");

现在我觉得奇怪的是，当我在调试中运行代码并在 Trigger 的 clear 函数中设置断点时，它会不断被调用。即使在触发器中没有达到 FIRE_AND_PURGE 点。所以感觉就像我完全误解了触发器应该如何工作。而且我的实现根本没有做我认为它正在做的事情。

我想我的问题是，触发器什么时候应该调用 clear ？这是实现组合 EventTimeTrigger 和 ProcessingTimeTrigger 的正确方法吗？

感谢我能得到的所有帮助。

更新 1：（2020-05-29）

为了提供更多关于如何设置的信息。 我的环境设置如下：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRestartStrategy(RestartStrategies.failureRateRestart(60, Time.of(60, TimeUnit.MINUTES), Time.of(60, TimeUnit.SECONDS)));
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

所以我对整个流使用 EventTime。 然后我像这样创建窗口：

DataStream<PlayerSession> playerSessions = sideEvents
                .keyBy((KeySelector<HashMap, String>) event -> (String) event.get(Field.SESSION_ID))
                .window(ProcessingTimeSessionWindows.withGap(org.apache.flink.streaming.api.windowing.time.Time.minutes(5)))
                .trigger(new EventTimeProcessingTimeTrigger(SESSION_TIMEOUT))
                .aggregate(new SessionAggregator())
                .name("Aggregate events into sessions")
                .uid("Aggregate events into sessions");

Answer 1

这种情况很复杂。我不敢确切地预测这段代码会做什么，但我可以解释一些正在发生的事情。

第 1 点：您已将时间特征设置为事件时间，安排了时间戳和水印，并在触发器中实现了 onEventTime 回调。但是您在任何地方都没有创建事件时间计时器。除非我错过了什么，否则实际上并没有使用事件时间或水印。您还没有实现事件时间触发器，我不希望 onEventTime 会被调用。

第 2 点：您的触发器不需要调用 clear。作为清除窗口的一部分，Flink 负责在触发器上调用 clear。

第 3 点：您的触发器试图反复触发和清除窗口，这似乎不正确。我这样说是因为您正在为每个元素创建一个新的处理时间计时器，并且当每个计时器触发时，您正在触发并清除窗口。您可以根据需要多次触发窗口，但您只能清除一次窗口，之后它就会消失。

第 4 点：会话窗口是一种特殊的窗口，称为 merging 窗口。当会话合并时（这种情况一直发生，随着事件的到来），它们的触发器被合并，其中一个被清除。这就是为什么你会经常看到 clear 被调用的原因。

建议：由于您有每分钟一次的保活，并且打算在 2 分钟不活动后关闭会话，因此您似乎可以将会话间隔设置为 2 分钟，这样可以避免一些正在发生的事情事情如此复杂。让会话窗口完成它们的设计目标。

假设这可行，那么您可以简单地扩展 Flink 的 ProcessingTimeTrigger 并覆盖其 onElement 方法来做到这一点：

@Override
public TriggerResult onElement(HashMap element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {

    if (endSession.contains(element.get(Field.EVENT_TYPE))) {
        return TriggerResult.FIRE_AND_PURGE;
    }

    return super(element, timestamp, window, ctx);
}

以这种方式，窗口将在两分钟不活动后触发，或者由明确的会话结束事件触发。

您应该能够简单地继承ProcessingTimeTrigger 的其余行为。

如果你想使用事件时间，那么使用EventTimeTrigger 作为超类，你必须找到一种方法来确保你的水印即使在流空闲时也能进行。请参阅this answer 了解如何处理。

Answer 2

同样的问题
我已将时间特性设置为处理时间和触发器：

//the trigger  

.trigger(PurgingTrigger.of(TimerTrigger.of(Time.seconds(winSec))))

以下触发函数：

//override the ProcessingTimeTrigger behavior
public class TimerTrigger<W extends Window> extends Trigger<Object, W> {
    private static final long serialVersionUID = 1L;
    private final long interval;
    private final ReducingStateDescriptor<Long> stateDesc;

    private TimerTrigger(long winInterValMills) { //window
        this.stateDesc = new ReducingStateDescriptor("fire-time", new TimerTrigger.Min(), LongSerializer.INSTANCE);
        this.interval = winInterValMills;
    }

    public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
        if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
            // if the watermark is already past the window fire immediately
            return TriggerResult.FIRE;
        }
        long now = System.currentTimeMillis();
        ReducingState<Long> fireTimestamp = (ReducingState) ctx.getPartitionedState(this.stateDesc);
        if (fireTimestamp.get() == null) {
            long time = Math.max(timestamp, window.maxTimestamp()) + interval;
            if (now-window.maxTimestamp()>interval){ // fire late
                time = (now-now%1000) + interval-1;
            }
            ctx.registerProcessingTimeTimer(time);
            fireTimestamp.add(time);
            return TriggerResult.CONTINUE;
        } else {
            return TriggerResult.CONTINUE;
        }
    }

    public TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception {
        if (time == window.maxTimestamp()){  
            return TriggerResult.FIRE;
        }
        return TriggerResult.CONTINUE;
    }

    public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
        ReducingState<Long> fireTimestamp = (ReducingState) ctx.getPartitionedState(this.stateDesc);
        if (((Long) fireTimestamp.get()).equals(time)) {
            fireTimestamp.clear();
            long maxTimestamp = Math.max(window.maxTimestamp(), time); //maybe useless
            if (maxTimestamp == time) {
                maxTimestamp = time + this.interval;
            }
            fireTimestamp.add(maxTimestamp);
            ctx.registerProcessingTimeTimer(maxTimestamp);
            return TriggerResult.FIRE;
        } else {
            return TriggerResult.CONTINUE;
        }
    }

    public void clear(W window, TriggerContext ctx) throws Exception {
        ReducingState<Long> fireTimestamp = (ReducingState) ctx.getPartitionedState(this.stateDesc);
        long timestamp = (Long) fireTimestamp.get();
        ctx.deleteProcessingTimeTimer(timestamp);
        fireTimestamp.clear();
    }

    public boolean canMerge() {
        return true;
    }

    public void onMerge(W window, OnMergeContext ctx) {
        ctx.mergePartitionedState(this.stateDesc);
    }

    @VisibleForTesting
    public long getInterval() {
        return this.interval;
    }

    public String toString() {
        return "TimerTrigger(" + this.interval + ")";
    }

    public static <W extends Window> TimerTrigger<W> of(Time interval) {
        return new TimerTrigger(interval.toMilliseconds());
    }

    private static class Min implements ReduceFunction<Long> {
        private static final long serialVersionUID = 1L;

        private Min() {
        }

        public Long reduce(Long value1, Long value2) throws Exception {
            return Math.min(value1, value2);
        }
    }
}