Apache Flink CEP中没有事件

Question

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

我试图检测的是，具有特定ID的CurrencyEvent类型的事件在一定时间内不会发生。我想每次在3000ms之后事件没有发生时检测没有这样的事件。

我的模式代码如下所示：

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

CEPPattern匹配一个或多个事件的序列；（间隔）子句中的添加了一个额外的约束，即序列中的所有事件都必须发生在指定的间隔内。当部分匹配超时时，这可以在TimedOutPARalMatchHandler中捕获。

在您的情况下，由于成功匹配的模式由单个事件组成，因此不可能存在部分匹配，并且匹配永远不会超时。（匹配序列的长度始终小于3秒。）

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

为了实现涉及缺失事件的用例比CEP提供的更大的灵活性，您可以将KeyedProcessFunction与定时器一起使用。