Phase 3: Data Pipelines & ETL

Flink windowing, watermarks & event-time processing

Advanced ~4 min read
Think of it this way A friendly analogy. Read this if the technical version feels dense. Show Hide

Imagine you’re watching a super fast basketball game where points are scored all the time, one after another, without stopping! You might want to know things like, “How many points did a team score in the first 10 minutes of the game?” or “What’s their average score over the last 5 minutes?” But since the game never really stops, how do you know when to stop counting for that "10 minutes" or "5 minutes"? You need a special way to chop up that continuous stream of points into smaller, manageable chunks. This is where the idea of "windows" comes in – they're like virtual buckets that collect events for a specific period.

Think of each basket being scored as an "event" in our game. Every time someone scores, it has a precise "game clock time" it happened – let's call this the event-time. It's the moment the ball actually went through the hoop. Now, imagine we have a super smart scoreboard that watches the game. For our "10-minute score", this scoreboard creates a window – a virtual bucket that collects all points that happened between 0:00 and 10:00 on the game clock. What if the scorekeeper is a bit slow and writes down a basket that happened at 5:00 game time, but doesn't record it until 6:00? The important thing is that our smart scoreboard still puts it in the 0:00-10:00 window because its event-time (5:00) falls there, not when it was recorded (6:00). This is called event-time processing: we care about when it actually happened in the game, not when our computer (or scorekeeper) saw it. This keeps our game statistics accurate and fair, no matter if some scores are written down a little late.

Sometimes, scores might arrive really late. If a basket from 5:00 game time only gets recorded at 15:00, our 0:00-10:00 window might have already closed! To deal with this, Flink uses something like a special "Game Time Announcer" called a watermark. The announcer periodically says, "Okay, we're pretty sure all scores that happened before 9:00 game time have now been seen. Any scores from before 9:00 that arrive after this announcement might be considered too late for that window." This helps Flink know when it's safe to close a window and give us the final score, even if a few points are a little delayed. Besides simple "10-minute buckets" (called Tumbling windows), we can also have "sliding windows" (like asking "What was the team's score in the last 5 minutes, updated every minute?") or "session windows" (where you count points for a player's "hot streak", and the window closes if they don't score for a while).

So, when people build big computer systems that track lots of fast-moving information, like monitoring website clicks, factory sensors, or even stock market trades, they use these ideas. This means you can accurately count how many users visited a certain page every hour, even if some users' clicks take a second or two longer to arrive. You can calculate the average temperature from a sensor every five minutes, knowing you're getting the true average based on when the temperature actually was that high, not just when the computer processed the reading. It helps make sure the information you get is always correct and reliable, just like a fair game score.

In stream processing, unbounded data streams necessitate boundaries for aggregations like counts or sums. Flink's windowing mechanism provides these boundaries, segmenting the continuous stream into finite chunks. Critically, for accurate and reproducible results in a distributed environment, Flink primarily relies on event-time processing. Unlike processing-time, which is when an event is processed by Flink (and is susceptible to network latency or backpressure), event-time refers to the timestamp embedded within the event itself, indicating when the event actually occurred. This ensures that aggregations are deterministic and correct, regardless of when data physically arrives in the system.

Flink supports various window types to cater to different analytical needs: Tumbling windows are fixed-size, non-overlapping windows (e.g., hourly sales counts); Sliding windows are fixed-size but overlap, providing moving aggregations (e.g., a 5-minute average updated every minute); and Session windows are dynamic, activity-based windows that close after a period of inactivity, ideal for user sessions. The challenge with event-time processing is handling out-of-order data arrival. This is where watermarks come in. A watermark is a special timestamp that Flink inserts into the stream, declaring that no more events with an event-time less than or equal to the watermark should be expected. They signal to Flink when it's safe to close a window and emit its results, providing a crucial balance between completeness and latency.

The synergy of event-time processing, flexible windowing, and intelligent watermarks forms the backbone of Flink's capability to deliver accurate and robust stream analytics. Watermarks allow Flink to correctly process late data up to a configurable 'allowed lateness' threshold, preventing windows from being held open indefinitely while still striving for accuracy. For a data engineer, mastering these concepts means building reliable, real-time data pipelines that produce consistent results, irrespective of the inherent complexities and unpredictability of distributed data streams.

Key Takeaways

  • Windowing segments unbounded data streams into finite chunks for aggregations.
  • Event-time processing ensures accurate, deterministic results based on when events occurred, not when they arrived.
  • Watermarks manage out-of-order data and signal when windows can be safely closed, balancing latency and completeness.
  • Flink offers Tumbling, Sliding, and Session windows for various aggregation patterns.
  • Combining these features enables robust and accurate real-time analytics in complex streaming environments.

Code Example

java
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.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

// Assume SensorReading is a POJO with fields: String id, long timestamp, double value
public class FlinkWindowingExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<SensorReading> sensorStream = env.fromElements(
            new SensorReading("sensor_1", 1678886400000L, 25.5), // March 15, 2023 12:00:00 AM UTC
            new SensorReading("sensor_1", 1678886403000L, 26.1),
            new SensorReading("sensor_2", 1678886401000L, 10.2),
            new SensorReading("sensor_1", 1678886408000L, 27.0)
        );

        // Assign timestamps and watermarks (allow 5 seconds out-of-orderness)
        DataStream<SensorReading> watermarkedStream = sensorStream.assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(5)) {
                @Override
                public long extractTimestamp(SensorReading element) {
                    return element.getTimestamp(); // Event timestamp is in milliseconds
                }
            }
        );

        // Apply a 10-second tumbling event-time window and sum sensor values by ID
        watermarkedStream
            .keyBy(s -> s.id)
            .window(TumblingEventTimeWindows.of(Time.seconds(10)))
            .sum("value")
            .print();

        env.execute("Flink Windowing Example");
    }
    
    // Dummy SensorReading class for example clarity
    public static class SensorReading {
        public String id; public long timestamp; public double value;
        public SensorReading() {} // Default constructor for Flink serialization
        public SensorReading(String id, long timestamp, double value) {
            this.id = id; this.timestamp = timestamp; this.value = value;
        }
        @Override public String toString() { return "SensorReading{" + "id='" + id + "', timestamp=" + timestamp + ", value=" + value + '}'; }
    }
}

How this code works

This Flink code demonstrates how to process a stream of sensor readings to calculate the sum of values within fixed time intervals, specifically for each sensor ID. It begins by setting up the StreamExecutionEnvironment and crucially specifies TimeCharacteristic.EventTime. This instructs Flink to use the timestamp embedded in each SensorReading event for all time-based operations, providing accurate results even if events arrive out of order. An initial DataStream of SensorReading objects is created, simulating a continuous stream of data from multiple sensors.

The assignTimestampsAndWatermarks step is central to event-time processing. It extracts the actual timestamp from each SensorReading and uses a BoundedOutOfOrdernessTimestampExtractor to generate watermarks. These watermarks are Flink's way of tracking event time progress, allowing it to determine when a window can be considered complete. A subtle but important detail is the Time.seconds(5) parameter: it creates a grace period, telling Flink to wait up to 5 seconds for late-arriving events before finalizing a window. The stream is then keyBy(s -> s.id) to group readings from the same sensor together, followed by window(TumblingEventTimeWindows.of(Time.seconds(10))) to define non-overlapping 10-second event-time windows. Finally, sum("value") aggregates the sensor values within each window, and the results are print()ed.

Resources

Related Articles

Hand-picked reads to go deeper on stream processing (kafka / flink).

Back to Data Engineer Roadmap