Sessionwindow of Flink

sessionWindows Session window ： Cut into different partition windows according to inactive time , And calculate the window

Sample environment

java.version: 1.8.x
flink.version: 1.11.1

Sample data source  （ Project code cloud download ）

Flink System examples And Build development environment and data

SessionWindow.java

import com.flink.examples.DataSource;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import java.util.List;

/**
 * @Description sessionWindows Session window ： Cut into different partition windows according to inactive time , And calculate the window 
 */
public class SessionWindow {

    /**
     *  Ergodic set , Return to the session sliding window after segmentation by inactive time , The largest age data record in the gender section under each window 
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // Set the flow processing time event , This time type must be set for the session window , There are three types :
        //1.ProcessingTime： With operator The processing time is subject to , It uses the system time of the machine as data stream Time for 
        //2.IngestionTime： Enter with data flink streaming data flow According to the time of 
        //3.EventTime： Subject to the time stamp field of the data , The application needs to specify how to start from record Extract the timestamp field from 
        env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
        env.setParallelism(4);
        DataStream<Tuple3<String, String, Integer>> inStream = env.addSource(new MyRichSourceFunction());
        DataStream<Tuple3<String, String, Integer>> dataStream = inStream.keyBy((KeySelector<Tuple3<String, String, Integer>, String>) k ->k.f1)
                // Press session window to scroll , When 2 No partition data stream specified in seconds , Then calculate once 
                // The session window is based on the fact that there is no active data access after a specified time , The window ends , Do window calculation 
                .window(EventTimeSessionWindows.withGap(Time.seconds(2)))
                .reduce(new ReduceFunction<Tuple3<String, String, Integer>>() {
                    @Override
                    public Tuple3<String, String, Integer> reduce(Tuple3<String, String, Integer> t1, Tuple3<String, String, Integer> t2) throws Exception {
                        // Return to the oldest 
                        return t1.f2 > t2.f2 ? t1: t2;
                    }
                });
        dataStream.print();
        env.execute("flink EventTimeSessionWindows job");
    }

    /**
     *  Analog data continues to output 
     */
    public static class MyRichSourceFunction extends RichParallelSourceFunction<Tuple3<String, String, Integer>> {
        @Override
        public void run(SourceContext<Tuple3<String, String, Integer>> ctx) throws Exception {
            List<Tuple3<String, String, Integer>> tuple3List = DataSource.getTuple3ToList();
            for (Tuple3 tuple3 : tuple3List){
                ctx.collect(tuple3);
                //1 One per second 
                Thread.sleep(2 * 1000);
            }
        }
        @Override
        public void cancel() {
            try{
                super.close();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}

Print the results

2> ( Zhang San ,man,20)
4> ( Li Si ,girl,24)
2> ( Wang Wu ,man,29)
4> ( Liu Liu ,girl,32)
2> ( Wu ba ,man,30)