Understanding Flink Window Types and Their Implementations

The article is part of a "Big Data Becoming God" series and offers a collection of links to introductory Flink material, DataSet/DataStream APIs, cluster deployment, restart strategies, and distributed cache.

Flink supports two primary ways to define windows: time‑based and count‑based. Each window has two key properties, size and interval. When size = interval a tumbling (non‑overlapping) window is formed; when size > interval a sliding (overlapping) window is created; if size < interval data loss occurs.

Combining these concepts yields four basic window types: time‑tumbling‑window (e.g., timeWindow(Time.seconds(5))), time‑sliding‑window (e.g., timeWindow(Time.seconds(5), Time.seconds(3))), count‑tumbling‑window (e.g., countWindow(5)), and count‑sliding‑window (e.g., countWindow(5,3)). Flink also allows windowing per key using keyBy.

Implementation examples are provided:

Tumbling Time Window

val counts: DataStream[(Int, Int)] = ...
val tumblingCnts: DataStream[(Int, Int)] = counts
  .keyBy(0)
  .timeWindow(Time.minutes(1))
  .sum(1)

Sliding Time Window

val slidingCnts: DataStream[(Int, Int)] = buyCnts
  .keyBy(0)
  .timeWindow(Time.minutes(1), Time.seconds(30))
  .sum(1)

Tumbling Count Window

val buyCnts: DataStream[(Int, Int)] = ...
val tumblingCnts: DataStream[(Int, Int)] = buyCnts
  .keyBy(0)
  .countWindow(100)
  .sum(1)

Session Window

val buyCnts: DataStream[(Int, Int)] = ...
val sessionCnts: DataStream[(Int, Int)] = buyCnts
  .keyBy(0)
  .window(ProcessingTimeSessionWindows.withGap(Time.seconds(30)))
  .sum(1)

In general, a window defines a finite set of elements on an infinite stream, which can be based on time, count, a combination of both, session gaps, or custom logic. Flink's DataStream API offers concise operators for common window operations while also supporting custom window assigners.

For the original article and additional resources, click the provided link.