Apache Flink Keyed Windows

Apache Flink windows operators allow to apply computations by splitting the stream of events into finite size buckets.

For a keyed stream, Flink allows to apply windowed computations using parallel tasks for each logical keyed stream.

Apache Flink window assigner uses to define how the elements are assigned to windows. Pre-defined window assigners can be identified as tumbling windows, sliding windows, sessions windows, and global windows.

For the below examples demonstration, let’s consider a stream of events of the Message class type.

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@AllArgsConstructor
@NoArgsConstructor
@Data
public class Message {
    private int id;
    private String reference;
    private int value;
}

Connect to the localhost port 1234 using netcat.

nc -l 1234

Define the keyed data stream using Message events.

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

SingleOutputStreamOperator<Message> sourceDataStream = env.socketTextStream("localhost", 1234)
        .map(
                value -> {
                    String[] values = value.split("\\s+");
                    return new Message(Integer.parseInt(values[0]), values[1], Integer.parseInt(values[2]));
                }
        );

org.apache.flink.streaming.api.datastream.KeyedStream<Message, String> messageDataStream = sourceDataStream.keyBy(Message::getReference);

Time-Driven Time Window

For a keyed stream time windows, desired window assigner can be declared with window( ).

public <W extends Window> WindowedStream<T, KEY, W> window(WindowAssigner<? super T, W> assigner)

• Tumbling Window

public static TumblingProcessingTimeWindows of(Duration size)

Example(s):

WindowedStream<Message, String, TimeWindow> windowedDataStream = messageDataStream.window(TumblingProcessingTimeWindows.of(Duration.ofSeconds(10)));

windowedDataStream.sum("value").print();

• Sliding Window

public static SlidingProcessingTimeWindows of(Duration size, Duration slide)

Example(s):

WindowedStream<Message, String, TimeWindow> windowedDataStream = messageDataStream.window(SlidingProcessingTimeWindows.of(Duration.ofSeconds(30), Duration.ofSeconds(5)));

windowedDataStream.sum("value").print();

• Session Window

public static ProcessingTimeSessionWindows withGap(Duration size)

Example(s):

WindowedStream<Message, String, TimeWindow> windowedDataStream = messageDataStream.window(ProcessingTimeSessionWindows.withGap(Duration.ofSeconds(10)));

windowedDataStream.sum("value").print();

• Global Window

public static GlobalWindows create()

Example(s):

WindowedStream<Message, String, GlobalWindow> windowedDataStream = messageDataStream.window(GlobalWindows.create());

windowedDataStream
        .trigger(CountTrigger.of(2))
        .sum("value")
        .print();

Data-Driven Count Window

• Tumbling Window

public WindowedStream<T, KEY, GlobalWindow> countWindow(long size)

Example(s):

WindowedStream<Message, String, GlobalWindow> windowedDataStream = messageDataStream.countWindow(3);

windowedDataStream.sum("value").print();

• Sliding Window

public WindowedStream<T, KEY, GlobalWindow> countWindow(long size, long slide)

Example(s):

WindowedStream<Message, String, GlobalWindow> windowedDataStream = messageDataStream.countWindow(3, 2);

windowedDataStream.sum("value").print();