在Apache Flink中动态添加模式而无需重新启动作业

我的用例是我想将不同的CEP模式应用于同一数据流. CEP模式会动态出现&我希望将它们添加到flink，而无需重新启动作业.尽管可以通过实现IterativeCondition的自定义类来处理所有条件，但我的主要问题是时间条件仅接受TimeWindow；无法处理.有什么方法可以根据输入元素设置传递给.within()的值?

My use case is that I want to apply different CEP patterns to the same datastream. the CEP patterns come dynamically & i want them to be added to flink without having to restart the job. While all conditions can be handled via custom classes that implement IterativeCondition, my main problem is that the temporal condition accepts only TimeWindow; which cannot be handled. Is there some way that the value passed to .within() be set based on the input elements?

有人问过类似的问题: Flink和动态模板识别

Something similar was asked here: Flink and Dynamic templates recognition

最佳答案: 可以添加的是一个共同平面映射运算符，该运算符可以在一个输入通道上接收事件，并在其他输入通道模式上接收事件.对于每个新接收的模式，要么更新现有的NFA(缺少此功能)，要么编译一个新的NFA.在后一种情况下，会将传入事件应用于所有存储的NFA."

Best Answer: "What one could add is a co-flat map operator which receives on one input channel the events and on the other input channel patterns. For each newly received pattern one either updates the existing NFA (this functionality is missing) or compiles a new one. In the latter case, one would apply incoming events to all stored NFAs."

我正在尝试实现这一目标，但是我遇到了一些困难.具体来说，在后一种情况下，会将传入事件应用于所有已存储的NFA"

I am trying to implement this but I am facing some difficulty. Specifically, on the point of "In the latter case, one would apply incoming events to all stored NFAs"

原因是我使用以下方法将流应用于模式:PatternStream matchStream = CEP.pattern(tmatchStream，pattern);

Reason being that I apply stream to pattern using: PatternStream matchStream = CEP.pattern(tmatchStream, pattern);

但是不会在co-flatMap内部定义流"tmatchStream".我在这里想念什么吗???任何帮助将不胜感激.

But the stream "tmatchStream" would not be defined inside the co-flatMap. Am I missing something here??? Any help would be greatly appreciated.