Flink 如何使用 CEP 处理过期的密钥

Question

我有一个监听事件的流式作业，使用 CEP 对它们进行操作。

流量是

stream = source
           .assignTimestampsAndWatermarks(...)
           .filter(...);

 CEP
   .pattern(stream.keysBy(e-> e.getId()), pattern)
   .process(PattenMatchProcessFunction)
   .sink(...);

键都是短暂的，进程函数不包含任何状态，也就是说可以通过设置ttl来移除状态。使用 EventTime 特征

我的问题，flink 如何处理过期的密钥，会对 GC 产生任何影响。 如果 flink 自己删除了密钥，那么这种情况发生的频率是多少。

面对 GC 问题，作业在部署 3 小时后卡住。 正在做内存调优，但想消除这种情况。

Answer 1

FsStateBackend 将为您的 CEP 操作员保存内存中的状态。

Flink 对 CEP 所做的是将元素缓冲在 MapState[Long, List[T]] 中，该 MapState[Long, List[T]] 将时间戳映射到当时到达的所有元素。一旦水印出现，Flink 会对缓冲的事件进行如下处理：

// 1) get the queue of pending elements for the key and the corresponding NFA,
// 2) process the pending elements in event time order and custom comparator if exists by feeding them in the NFA
// 3) advance the time to the current watermark, so that expired patterns are discarded.
// 4) update the stored state for the key, by only storing the new NFA and MapState iff they have state to be used later.
// 5) update the last seen watermark.

一旦事件处理完毕，Flink 将推进 watermark，这将导致状态中的旧条目过期（您可以在 NFA.advanceTime 中看到这一点）。这意味着驱逐您的元素取决于在您的流中创建和推送水印的频率。