Apache Flink-CoflatMapFunction不按事件时间顺序处理事件

Question

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

env.setParallelism(1)

DataStream<Integer> evenStream = env.addSource(new SourceFunction<Integer>(){
         public void run(SourceContext<Integer> ctxt){
                for (i=0; i < 20; i=i+2){
                     ctxt.collectWithTimestamp(i, i);
                     ctxt.emitWatermark(i);
                } 
         }
 }
 )
DataStream<Integer> oddStream  = env.addSource(new SourceFunction<Integer>(){
         public void run(SourceContext<Integer> ctxt){
                for (i=1; i < 21; i=i+2){
                     ctxt.collectWithTimestamp(i, i); // Using i as timestamp and watermark for this sample code, but in real code, I am using using timestamp of real event 
                     ctxt.emitWatermark(i);
                } 
         }
 }
 )

evenStream
   .connect(oddStream)
   .flatMap(new CoFlatMapFunction<Integer, Integer, Integer>(){

        public void flatMap1(Integer evenNumber, Collector<Integer> out){                  
               System.out.println(evenNumber);
        }
        public void flatMap2(Integer oddNumber, Collector<Integer> out){
               System.out.println(oddNumber);
        }

   }
   );

总之，我希望flatMap1()和flatMap2()按照我在事件中设置的时间戳的顺序被调用。但那不是真的。

Answer 1

Flink的协同函数（连接流上的函数）并不保证其方法的调用顺序。只要从两个输入中有可用的事件，就会调用这些方法（例如，FlatMap1()和FlatMap2())。在您的示例中，偶数源生成的数据量太小，因此当奇数到达时，所有数据都已处理完毕。

那么，事件时处理对于协同函数是如何工作的呢？

协函数的水印始终是两个输入的最小水印。对于coflatmapfunction来说，这其实并不重要，因为您既不能读取当前水印，也不能读取记录的时间戳。但是，通过coProcessFunction，您可以访问这两个选项，并且可以注册当水印到达某个时间点时调用的计时器。如果您想在事件时间上对流出的流进行排序，您需要缓冲传入的事件（处于状态），并且当水印进展时，您可以按顺序发出所有记录，直到超过水印的时间为止。