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

Question

我配置了事件时间处理并有一个带有CoFlatMapFunction 的连接流。我正在编写一个测试用例，但我看到 FlatMapFunction 没有使用事件时间顺序调用其方法 flatMap1() 和 flatMap2()。

一些伪代码需要澄清

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);
        }

   }
   );

当我运行它时，我希望它打印出来：

0,1,2,3,4....21

这是因为我交替设置偶数和奇数的时间戳。也就是说，0 的时间戳最低，其次是 1，其次是 2，以此类推。

但它首先打印所有偶数，然后是奇数。

总之，我希望按照我在事件中设置的时间戳顺序调用 flatMap1() 和 flatMap2()。但这并没有发生。

Answer 1

Flink 的协同函数（连接流上的函数）不保证它们的方法被调用的顺序。只要任一输入中的事件可用，就会调用方法（例如，flatMap1() 和 flatMap2()）。在您的示例中，偶数源生成的数据量太小，以至于当奇数到达时所有数据都已处理。

那么，事件时间处理如何为协同功能工作？

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