并行度>1时Flink CEP无法检测模式

Question

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.getConfig().setAutoWatermarkInterval(1000);
env.setParallelism(1); //tried with 1 & 4
.....    
DataStream<LogEvent> inputLogEventStream = env
    .readFile(format, FILEPATH, FileProcessingMode.PROCESS_CONTINUOUSLY, 1000)
    .map(new MapToLogEvents())              
    .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LogEvent>(Time.seconds(0)) {
            public long extractTimestamp(LogEvent element) {
                    return element.getTimeLong();
            }
    })
    .keyBy(new KeySelector<LogEvent, String>() {
            public String getKey(LogEvent le) throws Exception {
                    return le.getUser();
            }
    });
inputLogEventStream.print();

Pattern<LogEvent, ?> mflPattern = Pattern.<LogEvent> begin("mfl")
    .subtype(LogEvent.class).where(
            new SimpleCondition<LogEvent>() {
                    public boolean filter(LogEvent logEvent) {
                            if (logEvent.getResult().equalsIgnoreCase("failed")) { return true; }
                            return false;
                    }
    })
    .timesOrMore(3).within(Time.seconds(60));

PatternStream<LogEvent> mflPatternStream = CEP.pattern(inputLogEventStream, mflPattern);

DataStream<Threat> outputMflStream = mflPatternStream.select(
    new PatternSelectFunction<LogEvent, Threat>() {
            public Threat select(Map<String, List<LogEvent>> logEventsMap) {
                    return new Threat("MULTIPLE FAILED LOGINS detected!");
            }
    });
outputMflStream.print();

并行度=1（成功检测到模式）

04/03/2018 12:03:53 Source: Custom File Source(1/1) switched to RUNNING 
04/03/2018 12:03:53 SelectCepOperator -> Sink: Unnamed(1/1) switched to RUNNING 
04/03/2018 12:03:53 Split Reader: Custom File Source -> Map -> Timestamps/Watermarks(1/1) switched to RUNNING 
04/03/2018 12:03:53 Sink: Unnamed(1/1) switched to RUNNING 
LogEvent [recordType=base18, eventCategory=login, user=paul, machine=laptop1, result=failed, eventCount=1, dataBytes=100, time=2018-03-26T22:30:08Z, timeLong=1522103408000]
LogEvent [recordType=base19, eventCategory=login, user=deb, machine=desktop1, result=failed, eventCount=1, dataBytes=100, time=2018-03-26T22:30:03Z, timeLong=1522103403000]
LogEvent [recordType=base20, eventCategory=login, user=deb, machine=desktop1, result=failed, eventCount=1, dataBytes=100, time=2018-03-26T22:30:05Z, timeLong=1522103405000]
LogEvent [recordType=base21, eventCategory=login, user=deb, machine=desktop1, result=failed, eventCount=1, dataBytes=100, time=2018-03-26T22:30:06Z, timeLong=1522103406000]
**THREAT** ==> MULTIPLE FAILED LOGINS detected!

并行度=4（无法检测到模式）

04/03/2018 12:05:33 Split Reader: Custom File Source -> Map -> Timestamps/Watermarks(3/4) switched to RUNNING 
04/03/2018 12:05:33 Split Reader: Custom File Source -> Map -> Timestamps/Watermarks(2/4) switched to RUNNING 
04/03/2018 12:05:33 Sink: Unnamed(2/4) switched to RUNNING 
04/03/2018 12:05:33 SelectCepOperator -> Sink: Unnamed(2/4) switched to RUNNING 
04/03/2018 12:05:33 Sink: Unnamed(3/4) switched to RUNNING 
04/03/2018 12:05:33 SelectCepOperator -> Sink: Unnamed(3/4) switched to RUNNING 
2> LogEvent [recordType=base18, eventCategory=login, user=paul, machine=laptop1, result=failed, eventCount=1, dataBytes=100, time=2018-03-26T22:30:08Z, timeLong=1522103408000]
3> LogEvent [recordType=base21, eventCategory=login, user=deb, machine=desktop1, result=failed, eventCount=1, dataBytes=100, time=2018-03-26T22:30:06Z, timeLong=1522103406000]
3> LogEvent [recordType=base20, eventCategory=login, user=deb, machine=desktop1, result=failed, eventCount=1, dataBytes=100, time=2018-03-26T22:30:05Z, timeLong=1522103405000]
3> LogEvent [recordType=base19, eventCategory=login, user=deb, machine=desktop1, result=failed, eventCount=1, dataBytes=100, time=2018-03-26T22:30:03Z, timeLong=1522103403000]

Answer 1

我认为，在使用.keyby（）处理CEP时，不同的分区会得到这些事件，这是非常重要的。

您的代码

PatternStream<LogEvent> mflPatternStream = CEP.pattern(inputLogEventStream, mflPattern);

我相信应该

PatternStream<LogEvent> mflPatternStream = CEP.pattern(inputLogEventStream.keyBy("eventCategory","user"), mflPattern);