Apache Flink CEP 中没有事件

Question

我是 Apache Flink CEP 的新手，我正在努力检测一个简单的事件缺失。

我要检测的是具有特定 id 在特定时间内不会发生 CurrencyEvent 类型的事件。 每次 3000 毫秒后事件未发生时，我都想检测到此类事件的缺失。

我的模式代码如下：

Pattern<CurrencyEvent, ?> myPattern = Pattern.<Event>begin("CurrencyEvent")
        .subtype(CurrencyEvent.class)
        .where(new SimpleCondition<CurrencyEvent>() {
            @Override
            public boolean filter(CurrencyEvent currencyEvent) throws Exception {
                return currencyEvent.getId().equalsIgnoreCase("usd");
            }
        })
        .within(Time.milliseconds(3000L));

所以现在我的想法是使用超时函数来检测超时事件：

    DataStreamSource<Event> events = env.addSource(new TestSource(
            Arrays.asList(
                    basicCurrencyWithMivLevelEvent("EUR", 100L, Arrays.asList("1", "2"), 200D),
                    basicCurrencyWithMivLevelEvent("USD", 100L, Arrays.asList("1", "2"), 200D),
                    basicCurrencyWithMivLevelEvent("EUR", 100L, Arrays.asList("1", "2"), 200D)
            ),
            1636040364820L, // initial timestamp for the first element
            7000 // 7 seconds between each event
    ));

  PatternStream<Event> patternStream = CEP.pattern(
            events,
            (Pattern<Event, ?>) myPattern
    );

    OutputTag<Alarm> tag = new OutputTag<Alarm>("currency-timeout"){};

    PatternFlatTimeoutFunction<Event, Alarm> eventAlarmTimeoutPatternFunction = (patterns, timestamp, ctx) -> {
        System.out.println("New alarm, since after 3 seconds an event with id=usd is not detected");
        //TODO: call collect

    };


    PatternFlatSelectFunction<Event, Alarm> eventAlarmPatternSelectFunction = (patterns, ctx) -> {
        System.out.println("Select! (we can ignore it) " + patterns);
        // ignore matched events
    };

    return patternStream.flatSelect(
            tag,
            eventAlarmTimeoutPatternFunction,
            TypeInformation.of(Alarm.class),
            eventAlarmPatternSelectFunction
    );

我的测试源使用的是事件时间戳和水印，如下所示：

public class TestSource implements SourceFunction<Event> {

    private final List<Event> events;
    private final long initialTimestamp;
    private final long timeBetweenInMillis;

    public TestSource(List<Event> events, long initialTimestamp, long timeBetweenInMillis){
        this.events = events;
        this.initialTimestamp = initialTimestamp;
        this.timeBetweenInMillis = timeBetweenInMillis;


    }
    @Override
    public void run(SourceContext<Event> sourceContext) throws InterruptedException {
        long timestamp  = this.initialTimestamp;
        for(Event event: this.events){
            sourceContext.collectWithTimestamp(event, timestamp);
            sourceContext.emitWatermark(new Watermark(timestamp));
            timestamp+=this.timeBetweenInMillis;
        }
    }

    @Override
    public void cancel() {

    }

}

我正在使用 TimeCharacteristics.EventTime。

由于窗口时间（3 秒）低于每个事件之间的事件时间差（7 秒），我希望得到一些超时事件，但我得到的是 0。

Answer 1

CEP Pattern 匹配一个或多个事件的序列； within(interval) 子句添加了一个附加约束，即序列中的所有事件都必须在指定的间隔内发生。当部分匹配超时时，可以在TimedOutPartialMatchHandler 中捕获。

在您的情况下，由于成功匹配的模式由单个事件组成，因此不能有部分匹配，并且匹配永远不会超时。 （您的匹配序列总是少于 3 秒。）

您可以做的是扩展模式定义以包含第二个事件，以便匹配必须在 3 秒内有一个开始事件，然后是另一个事件。当第二个事件丢失时，您将有部分匹配超时。

要获得比 CEP 为实现涉及丢失事件的用例提供的更多灵活性，您可以将 KeyedProcessFunction 与计时器一起使用。