The most complete analysis of Flink frame window function

summary

In the real scene, the data flow often has no boundaries , Endless , It's like a channel in which water continues to flow elsewhere through a pipe , This is obviously impossible to deal with 、 Calculated , So how can we process data without boundaries ？ We can cut these unbounded data streams 、 Split , Get a bounded data set and process it 、 The calculation is much more convenient .

Flink Middle window （Window） Is to deal with unbounded data flow , Cut the wireless data stream into finite streams , Then the finite flow data after cutting is distributed to the bucket with specified finite size for analysis and calculation .

Window type

Flink There are two types of windows in ：

Time window （Time Window）

、

Count window （Count Window）

. The time window contains ： Scroll the time window （Tumbling Window）、 Slide the time window （Sliding Window）、 Session window （Session Window）. The count window contains ： Scroll count window and slide count window .

Scroll the window （Tumbling Windows）

• Take the time window as an example （ The counting window is similar to ）, Scrolling window is to segment data at fixed time intervals .

• The characteristic is that the time is relatively aligned 、 The length of windows is fixed and there is no overlap .

The scrolling window allocator allocates each element to a window of a specified window size , The scrolling window has a fixed size , And there's no overlap .

let me put it another way ： If you make a 30 Scrolling window at minute intervals , Then the unbounded data will be 30 Minute is a window period, which is cut into a limited data set . Applicable scenario ： Do statistical calculations . Do aggregate calculation for each time period .

The sliding window （Sliding Windows）

• Take the time window as an example （ The counting window is similar to ）, Sliding window is another form of fixed window , Sliding window consists of fixed window length and sliding interval .

• The window length is fixed , Windows can overlap .

explain ： The sliding window allocator allocates elements to fixed length windows , Similar to scrolling windows , The size of the window is configured by the window size parameter , Another window slide parameter controls how often the sliding window starts . therefore , Sliding window if the sliding parameter is smaller than the window size , Windows can overlap , In this case, the element is assigned to multiple windows .

Applicable scenario ：（ Ask for the latest interface  5min  To determine whether to call the police ） Statistics for the last period of time .

Session window （Session Windows）

• Session exposure only exists in the time window , Count window no session window .

• It is characterized by no alignment of time

Composed of a series of events to a specified length of time  timeout  Gap composition , Be similar to  web  Applied session, In other words, a new window will be generated if no new data is received for a period of time .

When it no longer receives elements in a fixed period of time , That is, inactive intervals produce , That window will close . One  session  The window passes through a  session  Interval to configure , This  session  The interval defines the length of the inactive period , When this inactive cycle occurs , So the current  session  Will be closed and subsequent elements will be assigned to the new  session  In the window

Window API Use

Window splitter window()

stay flink Can be used in the  

.window()

  To define a window , And then based on this  window  To do some aggregation or other processing operations . Be careful  

window ()

  Method must be in  

keyBy

  Then we can use it .

window()

  The input parameter received by the method is a  

WindowAssigner

 

WindowAssigner

  Responsible for distributing each input data to the correct  

window

  in .Flink  Provides general  WindowAssigner： Scroll the window （tumbling window）、 The sliding window （sliding window）、  Session window （session window）、 Global window （global window）

The code is as follows ：

public class WindowTest1_TimeWindow {
 public static void main(String[] args) throws Exception {
 // Build a stream processing environment
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 // Facility parallelism is 1
 env.setParallelism(1);

 DataStreamSource<String> inputStream = env.socketTextStream(&quot;localhost&quot;, 7777);


 DataStream<SensorReading> dataStream = inputStream.map(line -> {
 String[] fields = line.split(&quot;,&quot;);
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });

 // Windowing test   Specify the window allocator
 DataStream<Integer> resultStream = dataStream.keyBy(&quot;id&quot;)
 // Set up a 15 A scrolling window in seconds
 .window(TumblingProcessingTimeWindows.of(Time.seconds(15)));
 // Session window
 //.window(ProcessingTimeSessionWindows.withGap(Time.seconds(15)))
 // The sliding window
 //.window(SlidingProcessingTimeWindows.of(Time.seconds(20),Time.seconds(10)))
 
 env.execute();
 }
}


Flink  Provides a simpler  

.timeWindow

  and  

.countWindow

  Method , Used to define the time window and count window .

TimeWindow

TimeWindow  It is to combine all the data within the specified time range into a  window, One at a time window  All the data in it is calculated . The time interval can pass through  

Time.milliseconds(x)

,

Time.seconds(x)

,

Time.minutes(x)

Wait for one of them to specify .

CountWindow

CountWindow  According to the same in the window  key  The number of elements to trigger execution , At execution time, only the number of elements up to the window size is calculated  key  The corresponding result .

CountWindow  Of  window_size  It means the same  Key  The number of elements of , Not the total number of all elements entered .

Create different types of windows

Scroll the time window （tumbling time window）

.timeWindow(Time.seconds(15))


Slide the time window （sliding time window）

== In the following code  sliding_size  Set up in order to  5s, in other words , Every time  5s  Just calculate the output once , For each calculation  window  The scope is  15s  All the elements inside .==

 .timeWindow(Time.seconds(15),Time.seconds(5))


Session window （session window）

.window(ProcessingTimeSessionWindows.withGap(Time.seconds(15)))


Scroll the count window （tumbling count window）

== default  CountWindow  It's a scrolling window , Just specify the window size , When the number of elements reaches the window size , It will trigger the execution of the window .==

.countWindow(10)


Slide the count window （sliding count window）

== In the following code  sliding_size  Set up in order to  2, in other words , Every two received are the same  key  The data is calculated once , For each calculation  window  The scope is  10  Elements .==

.countWindow(10,2)


Window function

Flink Defines the calculation operations to be performed on the data collected in the window , It can be divided into two categories ： Incremental aggregate function 、 Full window functions .

• Incremental aggregate function ： Each piece of data comes and is calculated , Keep a state first , Aggregate functions have
  ReduceFunction
   
  AggregateFunction
  .

• 

Case code ：

public class WindowTest1_TimeWindow {
 public static void main(String[] args) throws Exception {
 // Build a stream processing environment
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 // Facility parallelism is 1
 env.setParallelism(1);
 
 DataStreamSource<String> inputStream = env.socketTextStream(&quot;localhost&quot;, 7777);


 DataStream<SensorReading> dataStream = inputStream.map(line -> {
 String[] fields = line.split(&quot;,&quot;);
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });

 // Windowing test   Specify the window allocator
 DataStream<Integer> resultStream = dataStream.keyBy(&quot;id&quot;)
 // Aggregate the window   Incremental window operation
 .timeWindow(Time.seconds(15))
 .aggregate(new AggregateFunction<SensorReading, Integer, Integer>() {
 @Override
 // Create accumulators
 public Integer createAccumulator() {
 return 0;
 }

 @Override
 public Integer add(SensorReading sensorReading, Integer accumulator) {
 return accumulator+1;
 }

 @Override
 public Integer getResult(Integer accumulator) {
 return accumulator;
 }

 @Override
 public Integer merge(Integer integer, Integer acc1) {
 return null;
 }
 });
 resultStream.print();
 env.execute();
 }
}


Local testing ： The result is output successfully .

• Full window functions ： First collect all the data in the window , When it's time to compute, it's going to traverse all the data . Corresponding function ：
  ProcessWindowFunction
  ,
  WindowFunction

Case code ：

public class WindowTest1_TimeWindow {
 public static void main(String[] args) throws Exception {
 // Build a stream processing environment
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 // Facility parallelism is 1
 env.setParallelism(1);



 DataStreamSource<String> inputStream = env.socketTextStream(&quot;localhost&quot;, 7777);


 DataStream<SensorReading> dataStream = inputStream.map(line -> {
 String[] fields = line.split(&quot;,&quot;);
 return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
 });
 
 // Full window function
 DataStream<Tuple3<String,Long,Integer>> resultStream = dataStream.keyBy(&quot;id&quot;)
 .timeWindow(Time.seconds(15))
 .apply(new WindowFunction<SensorReading, Tuple3<String,Long,Integer>, Tuple, TimeWindow>() {
 @Override
 public void apply(Tuple tuple, TimeWindow window, Iterable<SensorReading> input, Collector<Tuple3<String,Long,Integer>> out) throws Exception {
 String id =tuple.getField(0);
 Long windowEnd =window.getEnd();

 Integer count = IteratorUtils.toList(input.iterator()).size();
 out.collect(new Tuple3<>(id,windowEnd,count));
 }
 });
 
 resultStream.print();
 env.execute();
 }
}


Run tests locally ： Results output results

How to be in winodws Operating system nc Command to test the code ：

stay Windows How to use... In the operating system nc command

other API

• trigger ：
  .trigger()
    Definition  window  When to close , Trigger the calculation and output the result

• Remover ：
  .evitor()
    Define the logic for removing some data

• .allowedLateness()
    Allow processing of late data

• .sideOutputLateData()
    Put the late data into the side output stream

• .getSideOutput()
  Get side output stream

Case code ：

public class WindowTest2_CountWindow {
 public static void main(String[] args) throws Exception {
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 env.setParallelism(1);

 DataStreamSource<String> inputStream = env.socketTextStream(&quot;localhost&quot;, 7777);

 DataStream<SensorReading> dataStream = inputStream.map(line -> {
 String[] field = line.split(&quot;,&quot;);
 return new SensorReading(field[0], new Long(field[1]), new Double(field[2]));
 });

 //  Open counting window test
 DataStream<Double> resultStream = dataStream.keyBy(&quot;id&quot;)
 .countWindow(10, 2)
 .aggregate(new MyAvgTemp());

 // Other options API  Processing method of late data

 OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>(&quot;iate&quot;) {};

 SingleOutputStreamOperator<SensorReading> sumStream = dataStream.keyBy(&quot;id&quot;)
 .timeWindow(Time.seconds(15))
 //.trigger()
 //.evictor()
 .allowedLateness(Time.minutes(1))
 // Output to flow measurement
 .sideOutputLateData(outputTag)
 .sum(&quot;temperature&quot;);

 sumStream.getSideOutput(outputTag).print(&quot;late&quot;);

 resultStream.print();
 env.execute();
 }

 public static class MyAvgTemp implements AggregateFunction<SensorReading, Tuple2<Double,Integer>,Double>{

 @Override
 public Tuple2<Double, Integer> createAccumulator() {
 return new Tuple2<>(0.0,0);
 }

 @Override
 public Tuple2<Double, Integer> add(SensorReading sensorReading, Tuple2<Double, Integer> accumulator) {
 return new Tuple2<>(accumulator.f0+sensorReading.getTemperature(),accumulator.f1+1);
 }

 @Override
 public Double getResult(Tuple2<Double, Integer> accumulator) {
 return accumulator.f0 / accumulator.f1;
 }

 @Override
 public Tuple2<Double, Integer> merge(Tuple2<Double, Integer> a, Tuple2<Double, Integer> b) {
 return new Tuple2<>(a.f0+b.f0,a.f1+b.f1);
 }
 }
}


This is from the Internet ,window API  The overview

Be careful ：

• stay  flink In, we define a window, Must be in  keyBy After the operation .

• After the window function, there must be aggregation operation .