Definition
- Window – Groups an unbounded stream of events into finite buckets so that an aggregation function can run over each bucket.
- Tumbling window – Fixed-size, non-overlapping; each event belongs to exactly one window.
- Sliding window – Fixed-size, overlapping by a slide interval; one event can belong to multiple windows.
- Session window – Gap-based; closes after a period of inactivity, merging events into bursts.
- Global window – One window for all events; fires only when a custom trigger decides.
- Watermark – A special marker flowing with the stream asserting that event time has progressed to a certain point, computed as max-event-time minus out-of-orderness tolerance.
Imagine the National Bank of India wants to count how many transactions Rajesh makes every one minute. A tumbling window divides the day into back-to-back one-minute slots. A sliding window is like a security guard who every ten seconds looks back over the last minute. A session window groups a shopping session and closes when Rajesh goes quiet for five minutes. The watermark is the bank manager announcing "it is now safely 10:01, close the 10:00-to-10:01 slot and publish the count."
Understanding the Concepts
What are Flink windows?
Windows in Flink group an unbounded stream into finite buckets. There are two broad families: time-based windows (event time or processing time) and count-based windows (fixed number of elements). The four shapes are tumbling, sliding, session, and global.
import org.apache.flink.streaming.api.windowing.assigners.*;
import org.apache.flink.streaming.api.windowing.time.Time;
// Tumbling: 1-minute non-overlapping windows
stream.keyBy(txn -> txn.getAccountId())
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.sum("amount");
// Sliding: 1-minute window, 10-second slide
stream.keyBy(txn -> txn.getAccountId())
.window(SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(10)))
.sum("amount");
// Session: closes after 5-minute gap
stream.keyBy(txn -> txn.getAccountId())
.window(EventTimeSessionWindows.withGap(Time.minutes(5)))
.sum("amount");
// Global: one window, custom trigger
stream.keyBy(txn -> txn.getAccountId())
.window(GlobalWindows.create())
.trigger(CountTrigger.of(100))
.sum("amount");
How do tumbling windows work?
A tumbling window of size S produces non-overlapping slots. Each event belongs to exactly one window, determined by its event time divided by S. This is the most common window type for periodic aggregations.
// Per-account transaction count every 1 minute
DataStream<Tuple2<String, Long>> counts = transactions
.keyBy(txn -> txn.getAccountId())
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.apply(new WindowFunction<Transaction, Tuple2<String, Long>,
String, TimeWindow>() {
@Override
public void apply(String key, TimeWindow window,
Iterable<Transaction> input,
Collector<Tuple2<String, Long>> out) {
long count = 0;
for (Transaction t : input) count++;
out.collect(Tuple2.of(key, count));
}
});
Time: |--- 10:00-10:01 ---|--- 10:01-10:02 ---|--- 10:02-10:03 ---|
Events: | T1 T2 T3 T4 | T5 T6 | T7 T8 T9 |
Count: | 4 | 2 | 3 |
How do sliding windows work?
A sliding window has a size and a slide. A one-minute window sliding every ten seconds emits a view every ten seconds, and each event can belong to multiple windows. This gives a smoother, more frequent view at the cost of more computation.
// 1-minute window, sliding every 10 seconds
DataStream<Tuple2<String, BigDecimal>> sliding = transactions
.keyBy(txn -> txn.getAccountId())
.window(SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(10)))
.reduce((a, b) -> new Transaction(
a.getAccountId(), a.getAmount().add(b.getAmount())));
// Each event lands in size/slide = 60/10 = 6 windows
What is a watermark?
A watermark is a special marker that flows with the stream and asserts that event time has progressed to a certain point. It is computed as the maximum event time seen so far minus an out-of-orderness tolerance. When the watermark crosses a window's end time, the window fires.
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import java.time.Duration;
// Bounded out-of-orderness: watermark = max_seen - 5 seconds
DataStream<Transaction> withWatermarks = transactions
.assignTimestampsAndWatermarks(
WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(
Duration.ofSeconds(5))
.withTimestampAssigner(
(txn, ts) -> txn.getEventTimeMillis())
);
Events: T1(10:00:02) T2(10:00:05) T3(10:00:03) T4(10:00:08)
Max seen: 10:00:02 10:00:05 10:00:05 10:00:08
Watermark: 09:59:57 10:00:00 10:00:00 10:00:03
▲
Window [10:00:00, 10:00:01) fires
because watermark > window end
How do windows and watermarks work together?
The watermark drives window firing. A time-based window fires when the watermark passes the window end, meaning Flink believes all events for that window have arrived. Events that arrive after the watermark has passed their window are late; by default they are dropped.
// Window fires when watermark > window_end
withWatermarks
.keyBy(txn -> txn.getAccountId())
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.sum("amount");
// Window [10:00, 10:01) fires when watermark reaches 10:01
// Late event at 10:00:30 arriving after watermark passed 10:01 → dropped
What are the window components (assigner, trigger, evictor)?
Every window has three internal components. The assigner decides which window an element belongs to. The trigger decides when the window fires. The evictor optionally removes elements from the buffer.
// Custom trigger: fire early every 50 elements AND on watermark
stream.keyBy(txn -> txn.getAccountId())
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.trigger(CountTrigger.of(50)) // fire every 50 elements
.evictor(CountEvictor.of(10)) // keep only last 10 elements
.sum("amount");
How do you handle late events?
You can keep windows open with allowed lateness or route late events to a side output.
import org.apache.flink.util.OutputTag;
OutputTag<Transaction> lateTag = new OutputTag<>("late-txns") {};
SingleOutputStreamOperator<String> result = withWatermarks
.keyBy(txn -> txn.getAccountId())
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.minutes(5)) // keep window open 5 extra min
.sideOutputLateData(lateTag) // capture very late events
.apply(windowFunction);
// Retrieve late events for reconciliation
DataStream<Transaction> lateEvents = result.getSideOutput(lateTag);
What are common mistakes?
A common mistake is not setting a watermark, which means windows never fire because Flink has no assertion about time progress. Another mistake is using a very small out-of-orderness that drops too many slightly-delayed events.
// WRONG -- no watermark, windows never fire
stream.keyBy(txn -> txn.getAccountId())
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.sum("amount"); // Windows accumulate forever, never emit
// CORRECT -- watermark drives window firing
stream.assignTimestampsAndWatermarks(
WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(
Duration.ofSeconds(5))
.withTimestampAssigner((txn, ts) -> txn.getEventTimeMillis()))
.keyBy(txn -> txn.getAccountId())
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.sum("amount"); // Windows fire when watermark passes end
What should a senior engineer know?
The out-of-orderness tolerance trades latency against completeness. A larger value waits longer and catches more late events but adds delay, while a smaller value emits faster but drops more. For idle partitions, use withIdleness(Duration) so a stalled source does not hold back the global watermark and freeze all windows.
Visual Overview
Flink Window Types
Tumbling (1 min):
|-------|-------|-------|-------|
| W1 | W2 | W3 | W4 |
Each event in exactly one window
Sliding (1 min size, 10s slide):
|-------| |-------|
|-------| |-------|
|-------| |-------|
Events can be in multiple windows
Session (5 min gap):
|--events--| 5min gap |--events--|
| Session 1 | | Session 2 |
Closes after inactivity gap
Watermark driving window firing:
Events: ──T1──T2──T3──T4──T5──▶
Max ts: ─02──05──03──08──10──▶
WM(-5s): ─57──00──00──03──05──▶
▲ ▲
W[10:00,10:01) W[10:00,10:01)
NOT yet fired fires at WM=10:01
Key Takeaway
A Flink window turns an endless, unbounded stream into finite buckets that an aggregation can run over. There are four shapes: tumbling windows chop time into fixed, non-overlapping slots; sliding windows provide overlapping views that advance by a smaller slide; session windows group bursts of activity and close after a gap of inactivity; and global windows put everything into one window and rely on a custom trigger.
The piece that makes event-time windows correct is the watermark. Streams arrive out of order because of network delays and retries, so Flink needs a principled way to say "I have probably now seen everything up to time T." A watermark is exactly that assertion. The common strategy is bounded out of orderness, where the watermark is the maximum event time seen minus a tolerance. When the watermark crosses a window's end, the trigger fires that window.
Events that arrive after the watermark has passed their window are late. By default Flink drops them, but you can keep the window open with allowed lateness or route late events to a side output for separate handling. The out-of-orderness tolerance trades latency against completeness: a larger value waits longer and captures more stragglers but adds delay.
At the National Bank of India Rajesh used one-minute tumbling event-time windows to count transactions per account per minute for fraud detection, with a five-second watermark to absorb normal mobile-network delays. Genuinely late events were routed to a quarantine side output rather than dropped silently, because in banking you cannot lose a transaction.
Remember one sentence: A window slices the stream into finite buckets, and the watermark is the clock that tells each bucket when it is safe to close, fire, and emit its result.
---