java - FLINK CEP (Java 8) - 通过匹配模式持久化 "identity"

Question

我正在尝试使用 FLINK-CEP 来测量市场中的出价从 BidState.OPEN 到 BidState.Closed 所需的时间。我正在接收带有 ID 和状态的出价数据流，就目前情况而言，我将所有“开放”出价与所有“已关闭”出价进行匹配。

我在 patternStream.process 中有一个条件，它只允许将具有相同 ID 的开盘价和收盘价配对，正如它们应该的那样。但这感觉不对，因为匹配的数量以这种方式增长得非常快，而且我感觉这可以通过模式来完成。那么，有没有办法确保“开始”和“结束”对象具有相同的 ID？

AfterMatchSkipStrategy skipStrategy = AfterMatchSkipStrategy.noSkip();
//Is it possible to make sure that start.BidID == end.BidID in the pattern?
Pattern<BidEvent, ?> pattern = Pattern.<BidEvent>begin("start", skipStrategy).where(
        new SimpleCondition<BidEvent>() {
            @Override
            public boolean filter(BidEvent value) {
                return value.getState() == BidState.OPENED;
            }
        }).followedByAny("end").where(
        new SimpleCondition<BidEvent>() {
            @Override
            public boolean filter(BidEvent value) throws Exception {
                return value.getState() == BidState.CLOSED; // && value.getBidID == start.getBidID?
            }
        }).within(timeout);

PatternStream<BidEvent> patternStream = CEP.pattern(BidEventDataStream, pattern);

patternStream.process(new PatternProcessFunction<BidEvent, MatchingDuration>() {
    @Override
    public void processMatch(Map<String
            , List<BidEvent>> map
            , Context context
            , Collector<MatchingDuration> collector) {

        BidEvent start = map.get("start").get(0);
        BidEvent end = map.get("end").get(0);
        if (start.getBidId() == end.getBidId()){ // Make sure opening and closing bid is the same. Can this be done in the pattern?
            collector.collect(new MatchingDuration(start.getBidId(), (end.getTimestamp() - start.getTimestamp())));
        }
    }
}).addSink(matchingDurationSinkFunction);

Answer 1

我想出了如何获得我想要的行为:BidEventDataStream 必须键入，以便对具有相同属性的对象进行模式匹配 key 。问题中的代码无需更改，但必须编辑 BidEventDataStream 才能捕获 BidEvent.getBidId():

BidEventDataStream.keyBy(new KeySelector<BidEvent, Long>() {
                    @Override
                    public Long getKey(BidEventvalue) {
                        return value.getBidId();
                    }
                })