Apache Flink中BroadcastStream实现NOT IN逻辑的状态优化问题

需求背景 #

需要实现对应如下SQL的Flink作业：

SELECT *
  FROM EVENT_C c
 WHERE c.exclusion_id NOT IN (
    SELECT s.exclusion_id
      FROM EVENT_S s
      JOIN EVENT_A a
        ON s.input_id = a.input_id
     WHERE a.tags = 'some-value'
 )

当前已实现逻辑 #

已将每个Event作为独立DataStream处理，通过带时间窗口的JoinFunction完成EVENT_S与EVENT_A的关联，并尝试用BroadcastStream传递滚动窗口收集到的exclusion_id，再通过KeyedBroadcastProcessFunction处理EVENT_C记录与这些exclusion_id，核心代码如下：

关联并广播exclusion_id #

// Join EventS with EventA, broadcast the results
BroadcastStream<String> exclusionIdBroadcastStream = sStream
        .join(aStream)
        .where(new KeySelector<EventS, Object>() {
            @Override
            public Object getKey(EventS value) throws Exception {
                return value.getInputId();
            }
        })
        .equalTo(new KeySelector<EventA, Object>() {
            @Override
            public Object getKey(EventA value) throws Exception {
                return value.getInputId();
            }
        })
        .window(TumblingEventTimeWindows.of(Time.minutes(5)))
        .apply(new JoinFunction<EventS, EventA, String>() {
            @Override
            public String join(EventS first, EventA second) throws Exception {
                return first.getTurnId();
            }
        })
        .broadcast(turnIdStateDescriptor);

// Process EventC by performing a NOT IN with the broadcast ids
cStream
        .keyBy(new KeySelector<EventC, String>() {
            @Override
            public String getKey(EventC value) throws Exception {
                return value.getUniqueId();
            }
        })
        .connect(exclusionIdBroadcastStream)
        .process(new EventFilterBroadcastProcessFunction())
        .addSink(s3Sink);

KeyedBroadcastProcessFunction实现 #

public class EventFilterBroadcastProcessFunction extends KeyedBroadcastProcessFunction<String, EventC, String, OutputEvent> {

    private static final MapStateDescriptor<String, String> BROADCAST_DESCRIPTOR =
            new MapStateDescriptor<>("ExclusuionIdBroadcastState", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);

    private static final ListStateDescriptor<EventC> EVENT_C_DESCRIPTOR =
            new ListStateDescriptor<>("CustomerIdEvents", TypeInformation.of(EventC.class));

    private transient ListState<EventC> eventsToReprocessState;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        this.eventsToReprocessState = this.getRuntimeContext().getListState(EVENT_C_DESCRIPTOR);
    }

    @Override
    public void processElement(EventC event, ReadOnlyContext context, Collector<OutputEvent> out) throws Exception {
        final String turnId = context.getBroadcastState(BROADCAST_DESCRIPTOR).get(event.getTurnId());
        final long expirationTimestamp = nextTimeoutWindow(context.timestamp(), Time.minutes(1));

        if (turnId == null) {
            // Maybe the event is safe to collect, or the turnId has not been broadcast yet. Save it for later.
            this.eventsToReprocessState.add(event);

            context.timerService().registerEventTimeTimer(expirationTimestamp);
        }

    }

    @Override
    public void processBroadcastElement(String turnId, Context context, Collector<OutputEvent> out) throws Exception {
        context.getBroadcastState(BROADCAST_DESCRIPTOR).put(turnId, turnId);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<OutputEvent> out) throws Exception {
        super.onTimer(timestamp, ctx, out);
        int count = 0;
        int size = 0;
        final Iterable<EventC> eventsToReprocess = this.eventsToReprocessState.get();
        if (eventsToReprocess != null) {
            for (EventC event: eventsToReprocess) {
                size++;
                if (!ctx.getBroadcastState(BROADCAST_DESCRIPTOR).contains(event.getTurnId())) {
                    final OutputEvent OutputEvent = generateOutputEvent(event);
                    out.collect(OutputEvent);
                    count++;
                }
            }
            eventsToReprocessState.clear();
        }

    }

    private OutputEvent generateOutputEvent(final EventC event) {
        return OutputEvent.builder()
                ...
                .build();
    }

    public static long nextTimeoutWindow(final long timestamp, final Time interval) {
        return endOfWindow(timestamp, interval) + interval.toMilliseconds();
    }

    public static long beginningOfWindow(final long timestamp, final Time interval) {
        return timestamp - (timestamp % interval.toMilliseconds());
    }

    public static long endOfWindow(final long timestamp, final Time interval) {
        return beginningOfWindow(timestamp, interval) + interval.toMilliseconds();
    }
}

当前方案存在的问题 #

选择BroadcastStream是为了将exclusion_id视为规则，让所有并行实例基于同一规则集执行NOT IN以提升并行度，但Broadcast状态会持续增长无法删除元素，导致快照体积、CPU及内存占用不断上升。

疑问 #

• 是否可以在onTimer方法触发时清理Broadcast状态？
• 有没有更优的方案实现这类“关联/NOT IN”逻辑？

环境信息：Flink v1.11、Kinesis Data Analytics、JDK 11

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