Apache Flink: Stream Processing, Windows & State Management

Apache Flink is a true streaming engine that processes events one at a time with sub-millisecond latency, fault-tolerant managed state, and powerful windowing semantics. The framework of choice for fraud detection, real-time analytics, and event-driven architectures at scale.

What is Apache Flink?

Apache Flink is an open-source, distributed stream processing framework developed at TU Berlin and now one of the most active Apache projects. Unlike Spark Structured Streaming which processes data in micro-batches, Flink processes each event individually as it arrives, achieving consistent sub-millisecond latency. Flink provides exactly-once processing guarantees through distributed snapshots (Chandy-Lamport algorithm), making it reliable for financial and mission-critical applications.

Flink's key differentiator is managed state — stateful operations (running aggregates, session tracking, pattern matching) are stored in a configurable state backend and automatically checkpointed to durable storage. If a task fails, Flink restores state from the last checkpoint and replays from the source. Flink supports two primary APIs: the DataStream API for fine-grained control and Flink SQL for declarative streaming queries.

Window Types in Flink

• Tumbling Window — fixed-size, non-overlapping windows (e.g., 1-minute buckets); every event belongs to exactly one window
• Sliding Window — fixed-size, overlapping windows (e.g., 5-minute window every 1 minute); events can belong to multiple windows
• Session Window — dynamic gap-based windows; a window closes when no event arrives within the gap timeout; ideal for user session analysis
• Global Window — single unbounded window with a custom trigger; requires explicit trigger and evictor configuration

DataStream API with Kafka Source

# PyFlink DataStream API — event-time windowed aggregation
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer
from pyflink.common import WatermarkStrategy, Duration
from pyflink.common.serialization import SimpleStringSchema
from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.common.time import Time

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)

# Kafka source with event-time watermarks
source = KafkaSource.builder()     .set_bootstrap_servers("localhost:9092")     .set_topics("user-events")     .set_group_id("flink-consumer")     .set_starting_offsets(KafkaOffsetsInitializer.earliest())     .set_value_only_deserializer(SimpleStringSchema())     .build()

watermark_strategy = WatermarkStrategy     .for_bounded_out_of_orderness(Duration.of_seconds(5))     .with_timestamp_assigner(lambda event, _: extract_timestamp(event))

stream = env.from_source(source, watermark_strategy, "Kafka Source")

# Tumbling 1-minute window, keyed by user_id
result = (
    stream
    .map(lambda e: parse_event(e))          # (user_id, event_type, ts)
    .key_by(lambda e: e[0])                 # key by user_id
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .reduce(lambda a, b: (a[0], a[1], a[2] + b[2]))  # sum counts
)

result.print()
env.execute("User Event Aggregation")

Flink SQL Streaming Aggregations

-- Flink SQL — streaming join and windowed aggregation
-- Create Kafka source table
CREATE TABLE orders (
    order_id    STRING,
    user_id     STRING,
    amount      DOUBLE,
    event_time  TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic'     = 'orders',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format'    = 'json'
);

-- Create output sink (e.g., Iceberg table)
CREATE TABLE order_summary (
    window_start TIMESTAMP(3),
    window_end   TIMESTAMP(3),
    user_id      STRING,
    total_amount DOUBLE,
    order_count  BIGINT
) WITH (
    'connector' = 'iceberg',
    'catalog-name' = 'hive_catalog',
    'database-name' = 'analytics'
);

-- Tumbling window aggregation (Flink SQL TVF syntax)
INSERT INTO order_summary
SELECT
    window_start,
    window_end,
    user_id,
    SUM(amount)  AS total_amount,
    COUNT(*)     AS order_count
FROM TABLE(
    TUMBLE(TABLE orders, DESCRIPTOR(event_time), INTERVAL '1' MINUTE)
)
GROUP BY window_start, window_end, user_id;

Stateful Processing with KeyedProcessFunction

# Stateful processing with Flink ValueState
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.functions import KeyedProcessFunction
from pyflink.datastream.state import ValueStateDescriptor
from pyflink.common.typeinfo import Types

class FraudDetector(KeyedProcessFunction):
    """Flag users with >3 transactions in 60 seconds."""
    def open(self, runtime_context):
        desc = ValueStateDescriptor("tx_count", Types.INT())
        self.tx_count = runtime_context.get_state(desc)

    def process_element(self, value, ctx):
        current = self.tx_count.value() or 0
        current += 1
        self.tx_count.update(current)

        if current > 3:
            yield f"FRAUD ALERT: user {value['user_id']} tx count {current}"

        # Register a cleanup timer (TTL alternative)
        ctx.timer_service().register_event_time_timer(
            ctx.timestamp() + 60_000  # 60 seconds
        )

    def on_timer(self, timestamp, ctx):
        self.tx_count.clear()  # reset count after window

Flink vs Spark Structured Streaming vs Kafka Streams