WindowedStream聚合/Reduce操作的提前返回实现方案咨询

需求描述 #

需要对WindowedStream执行reduce操作，检测流中是否存在满足指定条件的元素：

• 若存在，输出该首个满足条件的元素；
• 找到后停止处理该key对应窗口的后续事件。

当前使用Flink 1.15.2，运行环境为Kinesis Data Analytics，现有简化Kotlin代码如下：

import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

data class Event(val eventId: Int, val timestampMillis: Long, val failed: Boolean)

fun eventAggregator(stream: DataStream<Event>): DataStream<Event> {
    // 使用事件自身的时间戳生成watermark
    val watermarks = WatermarkStrategy.forMonotonousTimestamps<Event>()
        .withTimestampAssigner { event, _ -> event.timestampMillis }
  
    // 按eventId分组
    val keyedStream = stream.assignTimestampsAndWatermarks(watermarks).keyBy(Event::eventId)

    // 1分钟滚动事件时间窗口
    val windowed = keyedStream.window(TumblingEventTimeWindows.of(Time.minutes(1)))
  
    // 逻辑：若eventId对应的事件中有失败的，标记该ID为失败
    return windowed.reduce { a, b -> if (a.failed) a else b }
}

尝试过的方案及问题 #

曾实现自定义Trigger，遇到失败事件时触发FIRE，否则委托给EventTimeTrigger，但该方式会触发整个窗口计算，导致多数其他事件结果错误。自定义Trigger代码如下：

import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger
import org.apache.flink.streaming.api.windowing.triggers.Trigger
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

class FireOnFirstFail() : Trigger<Event, TimeWindow>() {

    private val delegate = EventTimeTrigger.create()

    override fun onElement(element: Event, timestamp: Long, window: TimeWindow, ctx: TriggerContext) =
        if (element.failed) {
            TriggerResult.FIRE // 存在问题
        } else {
            delegate.onElement(element, timestamp, window, ctx)
        }
  
    override fun onProcessingTime(time: Long, window: TimeWindow, ctx: TriggerContext) =
        delegate.onProcessingTime(time, window, ctx)

    override fun onEventTime(time: Long, window: TimeWindow, ctx: TriggerContext) =
        delegate.onEventTime(time, window, ctx)

    override fun clear(window: TimeWindow, ctx: TriggerContext) =
        delegate.clear(window, ctx)
}

// 使用方式
// return windowed.trigger(FireOnFirstFail()).reduce...

解决方案 #

问题出在仅使用FIRE触发输出时，窗口状态并未被清除，后续该key的元素仍会被加入窗口处理，导致重复输出或错误结果。正确的做法是使用FIRE_AND_PURGE，它会在触发输出后立即清除窗口的所有状态，后续该窗口的元素将不再被处理。

修改后的自定义Trigger #

import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger
import org.apache.flink.streaming.api.windowing.triggers.Trigger
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

class FireOnFirstFail : Trigger<Event, TimeWindow>() {
    private val delegate = EventTimeTrigger.create()

    override fun onElement(element: Event, timestamp: Long, window: TimeWindow, ctx: TriggerContext): TriggerResult {
        return if (element.failed) {
            // 触发输出并清除窗口状态，后续该窗口不再接收元素
            TriggerResult.FIRE_AND_PURGE
        } else {
            delegate.onElement(element, timestamp, window, ctx)
        }
    }

    override fun onProcessingTime(time: Long, window: TimeWindow, ctx: TriggerContext): TriggerResult {
        return delegate.onProcessingTime(time, window, ctx)
    }

    override fun onEventTime(time: Long, window: TimeWindow, ctx: TriggerContext): TriggerResult {
        return delegate.onEventTime(time, window, ctx)
    }

    override fun clear(window: TimeWindow, ctx: TriggerContext) {
        delegate.clear(window, ctx)
    }
}

应用修改后的Trigger #

fun eventAggregator(stream: DataStream<Event>): DataStream<Event> {
    val watermarks = WatermarkStrategy.forMonotonousTimestamps<Event>()
        .withTimestampAssigner { event, _ -> event.timestampMillis }
  
    val keyedStream = stream.assignTimestampsAndWatermarks(watermarks).keyBy(Event::eventId)

    val windowed = keyedStream.window(TumblingEventTimeWindows.of(Time.minutes(1)))
  
    return windowed
        .trigger(FireOnFirstFail())
        .reduce { a, b -> 
            // 优先保留已失败的元素；若无失败元素，保留最后一个元素（可根据需求调整）
            if (a.failed) a else b 
        }
}

关键说明 #

1. FIRE_AND_PURGE的作用：触发窗口计算输出后，立即清除窗口的状态和缓存的元素，后续该key对应窗口的任何元素都会被忽略，完全停止处理。
2. reduce逻辑保留：原reduce逻辑会优先保留第一个出现的失败元素（因为一旦a是失败的，就直接返回a，不会再处理后续元素），符合“输出首个满足条件元素”的需求。
3. 无失败元素的处理：若窗口内没有失败元素，会在watermark超过窗口结束时间时，由EventTimeTrigger触发输出，返回窗口内最后一个非失败元素（可根据需求修改reduce逻辑为返回第一个元素）。

内容的提问来源于stack exchange，提问作者Jake