warnings.print（）打印以相反的顺序事件（最后的情况下第一）节选第一事件中的Apache弗林克CEP

我想筛选是>在弗林克10使用以下图案中的所有临时事件，warnings.print（）打印以相反的顺序事件（最后的情况下第一）节选第一事件中的Apache弗林克CEP

Pattern<MonitoringEvent, ?> warningPattern = Pattern.<MonitoringEvent>begin("first") 
       .subtype(TemperatureEvent.class) 
       .where(new FilterFunction<TemperatureEvent>() { 
        @Override 
        public boolean filter(TemperatureEvent temperatureEvent) throws Exception { 
         return temperatureEvent.getTemperature() > 50; 
        } 
       });

输入是一个文本文件，该文件被解析由输入函数以流，输入文件的内容是： -

这里第一值是Rack_id和第二是在温度

我已发出打印（）如下图所示

inputEventStream.print(); 
warnings.print();

现在两个输入流和WarnigsStream，自带问题，弗林克CEP的输出如下所示

08/10/2017 23:43:15 Job execution switched to status RUNNING. 
08/10/2017 23:43:15 Source: Custom Source -> Sink: Unnamed(1/1) switched to SCHEDULED 
08/10/2017 23:43:15 Source: Custom Source -> Sink: Unnamed(1/1) switched to DEPLOYING 
08/10/2017 23:43:15 AbstractCEPPatternOperator -> Map -> Sink: Unnamed(1/1) switched to SCHEDULED 
08/10/2017 23:43:15 AbstractCEPPatternOperator -> Map -> Sink: Unnamed(1/1) switched to DEPLOYING 
08/10/2017 23:43:15 AbstractCEPPatternOperator -> Map -> Sink: Unnamed(1/1) switched to RUNNING 
08/10/2017 23:43:15 Source: Custom Source -> Sink: Unnamed(1/1) switched to RUNNING 
Rack id = 1 and temprature = 98.0) 
Rack id = 2 and temprature = 33.0) 
Rack id = 3 and temprature = 44.0) 
Rack id = 4 and temprature = 55.0) 
Rack id = 5 and temprature = 66.0) 
Rack id = 6 and temprature = 88.0) 
Rack id = 7 and temprature = 99.0) 
Rack id = 8 and temprature = 76.0) 
08/10/2017 23:43:16 Source: Custom Source -> Sink: Unnamed(1/1) switched to FINISHED 
Rack id = 1 and temprature = 98.0) 
Rack id = 8 and temprature = 76.0) 
Rack id = 7 and temprature = 99.0) 
Rack id = 6 and temprature = 88.0) 
Rack id = 5 and temprature = 66.0) 
Rack id = 4 and temprature = 55.0) 
08/10/2017 23:43:16 AbstractCEPPatternOperator -> Map -> Sink: Unnamed(1/1) switched to FINISHED 
08/10/2017 23:43:16 Job execution switched to status FINISHED. 

Process finished with exit code 0

正如我们可以看到，第一复杂事件（架子ID = 1和温度= 98.0））以相同的顺序打印，但在此之后，具有temp> 50的所有其他复杂事件以与输入流相反的顺序打印。

My questions are :- 

1. Any idea why events are getting printed in reverse order? 
2. Is there a custom way to print values{w/o using warnings.print()} of 
    warning stream, like can I print only temperature, rather than rack-id ?

由于提前

来源

2017-08-11 Amarjit Singh

此问题是由分配时间戳和水印解决了的InputStream如下图所示

// Input stream of monitoring events 
     DataStream<MonitoringEvent> inputEventStream = env 
       .addSource(new InputStreamAGenerator()).assignTimestampsAndWatermarks(new IngestionTimeExtractor<>());

如下所示

08/11/2017 00:45:09 Job execution switched to status RUNNING. 
    08/11/2017 00:45:09 Source: Custom Source -> Timestamps/Watermarks(1/1) switched to SCHEDULED 
    08/11/2017 00:45:09 Source: Custom Source -> Timestamps/Watermarks(1/1) switched to DEPLOYING 
    08/11/2017 00:45:09 AbstractCEPPatternOperator -> Map -> Sink: Unnamed(1/1) switched to SCHEDULED 
    08/11/2017 00:45:09 AbstractCEPPatternOperator -> Map -> Sink: Unnamed(1/1) switched to DEPLOYING 
    08/11/2017 00:45:09 Source: Custom Source -> Timestamps/Watermarks(1/1) switched to RUNNING 
    08/11/2017 00:45:09 AbstractCEPPatternOperator -> Map -> Sink: Unnamed(1/1) switched to RUNNING 
    Rack id = 1 and temprature = 98.0) 
    Rack id = 4 and temprature = 55.0) 
    Rack id = 5 and temprature = 66.0) 
    Rack id = 6 and temprature = 88.0) 
    Rack id = 7 and temprature = 99.0) 
    Rack id = 8 and temprature = 76.0) 
    08/11/2017 00:45:10 Source: Custom Source -> Timestamps/Watermarks(1/1) switched to FINISHED 
    08/11/2017 00:45:10 AbstractCEPPatternOperator -> Map -> Sink: Unnamed(1/1) switched to FINISHED 
    08/11/2017 00:45:10 Job execution switched to status FINISHED.

生成的输出

来源

2017-08-11 04:48:08

warnings.print（）打印以相反的顺序事件（最后的情况下第一）节选第一事件中的Apache弗林克CEP

回答

相关问题