Stream Processing with Apache Flink

In this tutorial, you'll learn about Stream Processing with Apache Flink. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.

Apache Flink is a Stream Processing framework that processes data in real time with true record-at-a-time streaming, event-time semantics, stateful computations, and exactly-once consistency — unlike micro-batch architectures.

What You'll Learn

In this tutorial, you'll learn Flink's approach to Stream Processing — event time vs processing time, watermarks, stateful operators, Flink SQL, and CEP (Complex Event Processing) — with PyFlink code examples.

Why It Matters

Flink powers the most latency-sensitive use cases: fraud detection, algorithmic trading, and real-time personalization. Companies that need sub-millisecond latency and exactly-once semantics choose Flink over micro-batch systems.

Real-World Use

Alibaba uses Flink to Process trillions of events per day during Singles Day sales, handling 400,000+ orders per second with sub-second latency for real-time inventory updates, fraud detection, and personalized recommendations.

flowchart LR
  subgraph Sources
    A[Kafka]
    B[Kinesis]
    C[Files]
  end
  subgraph Flink Runtime
    D[JobManager]
    E[TaskManager 1]
    F[TaskManager 2]
    G[State Backend]
    H[Watermark Generator]
  end
  subgraph Sinks
    I[Kafka]
    J[Dashboard]
    K[Database]
  end
  A --> E
  B --> F
  E --> D
  F --> D
  E --> G
  F --> G
  E --> I
  F --> J
  E --> K

Event Time vs Processing Time

Understanding the time semantics is critical in Stream Processing.

Event time — When the event actually occurred (embedded in the data). This is the source of truth for most analytics.

Processing time — When the event is processed by Flink. This is simpler but inaccurate under backpressure.

Ingestion time — When the event enters Flink. A compromise between the two.

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.time_characteristic import TimeCharacteristic
from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.common.time import Time

env = StreamExecutionEnvironment.get_execution_environment()
env.set_stream_time_characteristic(TimeCharacteristic.EventTime)

def simulate_event_time_windowing():
    events = [
        ("sensor_1", 1.0, 1000),
        ("sensor_1", 2.0, 2000),
        ("sensor_1", 3.0, 3000),
        ("sensor_2", 1.5, 1500),
        ("sensor_2", 2.5, 2500),
    ]

    Windows = {}
    for sensor, value, event_time in events:
        window_start = (event_time // 2000) * 2000
        window_key = (sensor, window_start)
        if window_key not in Windows:
            Windows[window_key] = []
        Windows[window_key].append(value)

    for (sensor, window_start), values in sorted(Windows.items()):
        avg = sum(values) / len(values)
        print(f"Sensor: {sensor}, Window: {window_start}-{window_start+2000}, Avg: {avg:.2f}")

simulate_event_time_windowing()

Expected output:

Sensor: sensor_1, Window: 0-2000, Avg: 1.50
Sensor: sensor_1, Window: 2000-4000, Avg: 2.50
Sensor: sensor_2, Window: 0-2000, Avg: 1.50
Sensor: sensor_2, Window: 2000-4000, Avg: 2.50

Event-time Windows group events by when they happened, not when Flink processes them. This is essential for accurate analytics.

Watermarks for Handling Late Data

Watermarks track the progress of event time and determine when to trigger Windows. A watermark with value T means "no more events with event time < T will arrive."

def simulate_watermarked_windowing():
    events = [
        ("A", 1000),
        ("B", 1500),
        ("C", 2000),
        ("D", 2500),
        ("E", 1800), "# Late event
        ("F"", 3500),
    ]

    watermark = 0
    max_out_of_orderness = 1000
    window_size = 2000
    Windows = {}

    for event_id, event_time in events:
        watermark = max(watermark, event_time - max_out_of_orderness)
        window_start = (event_time // window_size) * window_size
        window_key = window_start

        if watermark >= window_start + window_size:
            print(f"ARRIVED LATE: {event_id} at time {event_time} (watermark={watermark})")
            continue

        if window_key not in Windows:
            Windows[window_key] = []
        Windows[window_key].append(event_id)
        print(f"Event {event_id} @{event_time} -> window {window_key}, watermark={watermark}")

    for w, evts in sorted(Windows.items()):
        print(f"\nWindow {w}-{w+window_size}: {evts}")

simulate_watermarked_windowing()

Expected output:

Event A @1000 -> window 0, watermark=0
Event B @1500 -> window 0, watermark=500
Event C @2000 -> window 0, watermark=1000
Event D @2500 -> window 2000, watermark=1500
Event E @1800 -> window 0, watermark=1500
Event F @3500 -> window 2000, watermark=2500

Window 0-2000: ['A', 'B', 'C']
Window 2000-4000: ['D', 'F']

Event E arrived late (at time 1800, after the watermark passed 1000) but was still within the allowed lateness, so it was included in window 0-2000.

Stateful Computations

Flink maintains State for operators across events. State can be keyed by a field (like user ID) and stored in RocksDB or in-memory backends.

def simulate_flink_State():
    """Simulate Flink's ValueState: track running count per key."""
    State = {}

    def Process_event(key, value):
        if key not in State:
            State[key] = {"count": 0, "sum": 0.0}
        State[key]["count"] += 1
        State[key]["sum"] += value
        avg = State[key]["sum"] / State[key]["count"]
        return avg

    events = [
        ("user_1", 100.0),
        ("user_2", 200.0),
        ("user_1", 150.0),
        ("user_1", 50.0),
        ("user_2", 300.0),
    ]

    for key, value in events:
        avg = Process_event(key, value)
        print(f"Event: {key}={value}, Running avg: {avg:.2f}")

simulate_flink_State()

Expected output:

Event: user_1=100.0, Running avg: 100.00
Event: user_2=200.0, Running avg: 200.00
Event: user_1=150.0, Running avg: 125.00
Event: user_1=50.0, Running avg: 100.00
Event: user_2=300.0, Running avg: 250.00

State is fault-tolerant in Flink. Checkpoints serialize the State to durable storage (DFS), enabling recovery from any point.

Flink SQL

Flink SQL allows you to run standard SQL on streaming data with the same semantics as batch SQL.

-- Flink SQL streaming query
CREATE TABLE orders (
  order_id BIGINT,
  user_id STRING,
  amount DECIMAL(10,2),
  order_time TIMESTAMP(3),
  WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'Kafka',
  'topic' = 'orders',
  'format' = 'JSON'
);

SELECT
  TUMBLE_END(order_time, INTERVAL '1' MINUTE) AS window_end,
  user_id,
  SUM(amount) AS total_spent,
  COUNT(*) AS order_count
FROM orders
GROUP BY
  TUMBLE(order_time, INTERVAL '1' MINUTE),
  user_id;

This query computes per-user spending totals every minute using event-time tumbling Windows with a 5-second allowed lateness.

CEP (Complex Event Processing)

Flink CEP detects patterns across event streams — useful for fraud detection and monitoring.

Common Mistakes Beginners Make

1. Confusing event time with processing time

Always use event time for accurate analytics. Processing time is only suitable for simple monitoring where exact timing doesn't matter.

2. Setting watermarks too aggressively

A watermark that assumes zero lateness drops valid late events. Set maxOutOfOrderness based on your actual data latency distribution.

3. Ignoring State size

Unbounded State grows forever. Configure State TTL (time-to-live) to expire old State and prevent memory exhaustion.

4. Using session Windows without understanding timeout

Session Windows Merge adjacent events with a gap. If the gap is too small, sessions split. If too large, Windows never close.

5. Not checkpointing frequently enough

Checkpoints enable recovery. Set checkpoint interval based on your recovery time objective (e.g., every 10 seconds for 10-second recovery).

Practice Questions

1. What is the difference between event time and processing time in Flink? Event time is when the event occurred (embedded in the data). Processing time is when Flink processes it. Event time is accurate but requires watermark handling.

2. What is a watermark and how does it work? A watermark tracks event-time progress. It signals that no events with a timestamp below the watermark value should arrive. Flink uses watermarks to trigger window computations.

3. How does Flink achieve exactly-once State consistency? Through distributed snapshots (checkpoints) using the Chandy-Lamport algorithm. On failure, Flink restores State from the last successful checkpoint and replays source data.

Challenge

Design a fraud detection pipeline with Flink CEP that detects: multiple failed logins followed by a successful login within 60 seconds, using event time and a 30-second watermark delay.

Real-World Task

Use PyFlink to read from a Kafka topic, apply a tumbling event-time window of 5 minutes, and write the aggregation results to a PostgreSQL database. Monitor the watermark progress.

FAQ

Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.

What's Next

Congratulations on completing this Apache Flink tutorial! Here's where to Go from here:

• Practice daily — Consistency is more important than long study sessions
• Build a project — Apply what you learned by building something real
• Explore related topics — Check out other tutorials in the same category
• Join the community — Discuss with other learners and share your progress

Remember: every expert was once a beginner. Keep coding!