Flink固定时长Session Window业务场景需求实现问题咨询

你要实现的是按key独立计时、起始时间对齐到每个key首个元素到达时间、长度固定为1分钟的滚动窗口，原生的固定时长Session Window本身的设计逻辑就是有新元素接入就延长窗口结束时间，和你的需求完全不匹配，无法通过调整Session Gap参数实现，以下是两种可行的落地方案：

方案1：使用KeyedProcessFunction实现（推荐） #

该方案逻辑清晰易维护，无需适配Flink窗口API的内置规则，灵活度更高。

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import com.google.common.collect.Lists;
import java.util.List;

// 替换泛型：KEY_TYPE为你的key类型，IN_TYPE为输入数据类型，OUT_TYPE为输出结果类型
public class KeyAlignedTumblingWindowProcessFunc<KEY_TYPE, IN_TYPE, OUT_TYPE> extends KeyedProcessFunction<KEY_TYPE, IN_TYPE, OUT_TYPE> {
    private transient ValueState<Long> windowEndTimeState;
    private transient ListState<IN_TYPE> windowElementsState;
    // 固定窗口时长1分钟，单位ms
    private static final long WINDOW_SIZE_MS = 60 * 1000L;
    // 可选：允许迟到时长，事件时间场景使用
    private static final long ALLOWED_LATENESS_MS = 0L;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Long> endTimeDesc = new ValueStateDescriptor<>("windowEndTime", Long.class);
        windowEndTimeState = getRuntimeContext().getState(endTimeDesc);
        ListStateDescriptor<IN_TYPE> elementsDesc = new ListStateDescriptor<>("windowElements", IN_TYPE.class);
        windowElementsState = getRuntimeContext().getListState(elementsDesc);
    }

    @Override
    public void processElement(IN_TYPE value, Context ctx, Collector<OUT_TYPE> out) throws Exception {
        Long endTime = windowEndTimeState.value();
        // 该key首次接入数据，初始化窗口
        if (endTime == null) {
            // 事件时间场景用这行获取时间戳，需提前配置Watermark生成规则
            long time = ctx.timestamp() != null ? ctx.timestamp() : 0L;
            // 处理时间场景用这行替换上面的time赋值
            // long time = System.currentTimeMillis();
            endTime = time + WINDOW_SIZE_MS;
            windowEndTimeState.update(endTime);
            // 注册触发定时器，事件时间用registerEventTimeTimer，处理时间用registerProcessingTimeTimer
            ctx.timerService().registerEventTimeTimer(endTime + ALLOWED_LATENESS_MS);
        }
        // 数据存入窗口状态
        windowElementsState.add(value);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT_TYPE> out) throws Exception {
        // 窗口触发，执行计算逻辑
        List<IN_TYPE> elements = Lists.newArrayList(windowElementsState.get());
        // 替换为你自己的窗口计算逻辑
        OUT_TYPE result = yourCalculateMethod(elements);
        out.collect(result);
        // 清空状态，等待该key下一个窗口的首个元素接入
        windowEndTimeState.clear();
        windowElementsState.clear();
    }

    // 替换为实际的窗口计算逻辑
    private OUT_TYPE yourCalculateMethod(List<IN_TYPE> elements) {
        return null;
    }
}

使用时直接在keyBy之后调用该函数即可：

dataStream
    .keyBy(/* 你的key选择器 */)
    .process(new KeyAlignedTumblingWindowProcessFunc<>())
    /* 后续输出逻辑 */;

方案2：自定义WindowAssigner实现 #

如果必须基于Flink官方窗口API体系开发，可以自定义窗口分配器，核心逻辑和方案1一致，需要关联keyed状态存储每个key的窗口起始时间，实现复杂度更高，灵活度低于方案1，仅适合有窗口API强依赖的场景。

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import java.time.Duration;
import java.util.Collections;
import java.util.Collection;

public class KeyAlignedTumblingWindowAssigner extends WindowAssigner<Object, TimeWindow> {
    private final long windowSizeMs;
    private final boolean isEventTime;

    private KeyAlignedTumblingWindowAssigner(long windowSizeMs, boolean isEventTime) {
        this.windowSizeMs = windowSizeMs;
        this.isEventTime = isEventTime;
    }

    public static KeyAlignedTumblingWindowAssigner of(Duration windowSize, boolean isEventTime) {
        return new KeyAlignedTumblingWindowAssigner(windowSize.toMillis(), isEventTime);
    }

    @Override
    public Collection<TimeWindow> assignWindows(Object element, long timestamp, WindowAssignerContext context) {
        // 此处需结合keyed状态读取当前key已存在的窗口，无匹配窗口则新建1分钟长度的窗口
        // 状态逻辑和方案1完全一致，仅套入窗口API结构
        long currentTime = isEventTime ? timestamp : context.getCurrentProcessingTime();
        long windowStart = getCurrentKeyWindowStart(currentTime); // 从状态读取对应key的窗口起始时间，无则取currentTime
        return Collections.singletonList(new TimeWindow(windowStart, windowStart + windowSizeMs));
    }

    @Override
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return isEventTime ? EventTimeTrigger.create() : ProcessingTimeTrigger.create();
    }

    @Override
    public TypeSerializer<TimeWindow> getWindowSerializer(ExecutionConfig executionConfig) {
        return new TimeWindow.Serializer();
    }

    @Override
    public boolean isEventTime() {
        return isEventTime;
    }

    // 实现从状态读取当前key窗口起始时间的逻辑
    private long getCurrentKeyWindowStart(long currentTime) {
        // 此处补充状态读写逻辑即可
        return currentTime;
    }
}

使用方式和普通窗口一致：

dataStream
    .keyBy(/* 你的key选择器 */)
    .window(KeyAlignedTumblingWindowAssigner.of(Duration.ofMinutes(1), /* 是否事件时间 */ true))
    .apply(/* 你的窗口计算函数 */)
    /* 后续输出逻辑 */;

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