Flink window mechanism (two waits and the last explanation)

1. Source code

package demo.flink.test;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

public class TestWindow {
    

    /** * *  Window mechanism  ( Two waits ,  Finally, find out the bottom ) * 1. WaterMark  Waterline . The window can set a short waiting time , When the following data arrives , Then close the window  * 2. allowLateness  Delay window closing time . *  Set a delay time after the window closes , Data arriving within the delay time , Window aggregation will be performed again during subsequent window calculation . * 3. sideOutputStream  Side output stream   This is the final plan . After the window completes the aggregation calculation , No longer receive data . *  These long overdue data , The user can only choose to collect another side output stream , Decide for yourself what to do with . * * @throws Exception */
    public static void main(String[] args) throws Exception {
    
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        //  Be careful :  Do not use fromElements,  Can't test the effect of the window 
// DataStreamSource<Integer> source = env.fromElements(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 1);

        /** * 1.  Set the following scenario  * 1) watermark  Delay 1 second  * 2) allowLateness  Delay 3 second  * 3) sideOutputStream  The bottom line  * * 2.  Try the following parameters  * 0, * 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 //  window 0  data  [1 ~ 11) * * 11 //  window 1  data  [11 ~ 21),  window 0  close  * 1, 2, 3 //  window 0  Delay data (1 second ) * * 12 //  window 1  data  [11 ~ 21) * 1, 2, 3 //  window 0  Delay data (2 second ) * * 13 //  window 1  data  [11 ~ 21) * 1, 2, 3 //  window 0  Delay data (3 second ) * * 14 //  window 1  data  [11 ~ 21) * 1, 2, 3 //  window 0  Delay data ( exceed 3 second ),  go sideOutputStream The underlying logic  * * 20 //  window 1  data  [11 ~ 21) * 21 //  window 2  data  [21, 31),  window 1  close  * 22, 23, 24, 25, 26, 27, 28, 29, 30 //  window 2  data  [21, 31) * 31 //  window 3  Turn on  * */
        SingleOutputStreamOperator<String> source = env.socketTextStream("localhost", 7777);

        OutputTag<Tuple1<Integer>> sideTag = new OutputTag<Tuple1<Integer>>("side"){
    };

        SingleOutputStreamOperator<Tuple1<Integer>> stream = source
                .map(Integer::parseInt)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Integer>forBoundedOutOfOrderness(Duration.ofSeconds(1)) // watermark  Delay 1 second 
                                .withTimestampAssigner((SerializableTimestampAssigner<Integer>) (element, recordTimestamp) -> element * 1000) // event time  Set up 
                )
                .map((MapFunction<Integer, Tuple1<Integer>>) value -> new Tuple1<Integer>(1))
                .returns(new TypeHint<Tuple1<Integer>>() {
    })
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10))) //  Scroll the window  10 second 
                .allowedLateness(Time.seconds(3)) // allowLateness  Delay 3 second 
                .sideOutputLateData(sideTag) // sideOutputStream  The bottom line 
                .sum(0);

        source.print("source-----------------");
        stream.print("stream-----------------");
        stream.getSideOutput(sideTag).print("side-----------------");

        env.execute("demo");
    }
}

2. Console output

nc -lk 7777 ##  monitor 7777 port 

##  Console output 
source-----------------> 0
source-----------------> 1
source-----------------> 2
source-----------------> 3
source-----------------> 4
source-----------------> 5
source-----------------> 6
source-----------------> 7
source-----------------> 8
source-----------------> 9
source-----------------> 10
source-----------------> 11
stream-----------------> (10)
source-----------------> 1
stream-----------------> (11)
source-----------------> 2
stream-----------------> (12)
source-----------------> 3
stream-----------------> (13)
source-----------------> 12
source-----------------> 1
stream-----------------> (14)
source-----------------> 2
stream-----------------> (15)
source-----------------> 3
stream-----------------> (16)
source-----------------> 13
source-----------------> 1
stream-----------------> (17)
source-----------------> 2
stream-----------------> (18)
source-----------------> 3
stream-----------------> (19)
source-----------------> 14
source-----------------> 1
side-----------------> (1)
source-----------------> 2
side-----------------> (1)
source-----------------> 3
side-----------------> (1)
source-----------------> 20
source-----------------> 21
stream-----------------> (5)
source-----------------> 22
source-----------------> 23
source-----------------> 24
source-----------------> 25
source-----------------> 26
source-----------------> 27
source-----------------> 28
source-----------------> 29
source-----------------> 30
source-----------------> 31
stream-----------------> (10)

3. Reference resources

• Flink Windows