Stream processing: Flink and Spark Structured Streaming

The previous lesson showed how Kafka stores a stream of events. A stream on its own answers no questions. To get "orders per 5 minutes", "is this payment suspicious?" or "which items are selling out right now" you need a stream processor: a program that reads events continuously, keeps some state and produces results as it goes.

This lesson explains the ideas every stream processor shares (event time, windows, watermarks, state, checkpoints) and compares the two engines you will meet most: Apache Flink and Spark Structured Streaming. You do not need to run anything, but the code samples show how it looks.

A batch job has a start and an end: it reads a finished set of data and stops. A streaming job never finishes, because the input never ends. That one difference causes new problems: how do you compute a total over an endless stream? What if an event arrives late? What happens if the job crashes after running for three weeks?

Some definitions help. Latency is how long between an event happening and the result being available. Batch pipelines accept minutes to hours; streaming aims for seconds or less. Most teams do not need real-time everywhere. Streaming is more complex and costly to run than batch, so ask "what decision gets better if this is 1 minute old instead of 1 day old?" before choosing it. A daily sales report is a good batch job. A live kitchen display or a fraud check is not.

Every event has two times. Event time is when it actually happened (the customer tapped Pay at 18:04:50). Processing time is when your system sees it (18:06:10, because the phone was on a poor Tier-3 mobile network). Events arrive late and out of order, especially from mobile apps.

If you group by processing time, a delayed order lands in the wrong minute and your numbers drift. If you group by event time, your answers match reality, but you must decide how long to wait for stragglers. Modern engines use event time as the default for analytics.

Infinite data is cut into finite chunks called windows, and you compute a result per window.

If events can arrive late, when is the 18:00-18:05 window finished? You can never be sure. A watermark is the engine's estimate of "we have probably seen all events up to time T". A typical rule is: watermark = the latest event time seen minus an allowed lateness, say 10 minutes. When the watermark passes the end of a window, the engine finalises that window, frees its state and treats later events for it as too late (they are dropped or sent to a side output, depending on the engine and settings).

The allowed lateness is a trade-off. A long wait gives more accurate results but higher latency and more memory. A short wait gives faster results but drops more late events. There is no perfect value; choose from how late your data actually arrives.

To count orders per window, the engine must remember running totals between events. This memory is called state. State is why stream processing is hard: it must survive crashes and restarts, and it can grow without bound if you are careless (for example, keeping every customer ever seen).

Engines handle failure with checkpoints: they periodically save their state and their position in the input (for Kafka, the offsets) to durable storage. After a crash they restore the last checkpoint and replay from that position. Because replay may repeat some events, engines combine checkpoints with idempotent or transactional writes to give exactly-once results in the output sink in many setups. Read the documentation of both your source and your sink to confirm what is guaranteed; "exactly-once" is always about a specific end-to-end setup, not magic.

Spark Structured Streaming lets you write a streaming job with the same DataFrame API you used in *Apache Spark and PySpark basics*. You treat the stream as a table that keeps growing and write a query on it. By default Spark processes data in small micro-batches (a tiny batch job every short interval), which gives latencies of around a second or more and reuses all of Spark's batch engine. Newer Spark versions also offer a lower-latency continuous mode; check the current documentation for its status and limits.

This example reads order events from a Kafka topic named order_events and counts units per item in 5-minute event-time windows, accepting events up to 10 minutes late:

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType

spark = SparkSession.builder.appName("live-orders").getOrCreate()

schema = StructType([
    StructField("order_id", StringType()),
    StructField("item_id", StringType()),
    StructField("qty", IntegerType()),
    StructField("event_time", TimestampType()),
])

raw = (
    spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "order_events")
    .option("startingOffsets", "earliest")
    .load()
)

orders = (
    raw.select(F.from_json(F.col("value").cast("string"), schema).alias("e"))
    .select("e.*")
)

units_per_window = (
    orders
    .withWatermark("event_time", "10 minutes")
    .groupBy(F.window("event_time", "5 minutes"), "item_id")
    .agg(F.sum("qty").alias("units"))
)

query = (
    units_per_window.writeStream
    .outputMode("update")
    .format("console")
    .option("checkpointLocation", "checkpoints/live-orders")
    .start()
)
query.awaitTermination()

Run with spark-submit and the Kafka connector package, whose version must match your Spark and Scala version: spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.13:<spark-version> live_orders.py. Check the Spark Structured Streaming + Kafka Integration guide for the exact coordinates (use _2.12 if your Spark build uses Scala 2.12).

Read the code in four parts. The source is readStream from Kafka, which gives a binary value column. The parse step casts it to a string and uses from_json with a schema. The logic sets a watermark of 10 minutes, then groups by a 5-minute tumbling window and item. The sink is writeStream; here it prints to the console, in update mode (only changed rows each batch). The checkpointLocation is where Spark saves offsets and state so the job can restart without losing its place.

Output modes matter: append writes a row only once it is final (for windowed aggregations, after the watermark passes), update writes changed rows, and complete rewrites the whole result each time (only practical for small results).

Apache Flink was built for streaming first. It processes events one at a time as they arrive (true record-at-a-time streaming), which gives very low latency, and it has fine-grained control over state, time and timers. It offers a SQL interface, a Table API and a lower-level DataStream API in Java, Python (PyFlink) and others. Flink also handles batch by treating it as a bounded stream.

The same 5-minute count in Flink SQL, run in the Flink SQL client with the Kafka connector available:

CREATE TABLE order_events (
    order_id   STRING,
    item_id    STRING,
    qty        INT,
    event_time TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '10' MINUTE
) WITH (
    'connector' = 'kafka',
    'topic' = 'order_events',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'flink-demo',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
);

SELECT window_start, window_end, item_id, SUM(qty) AS units
FROM TABLE(
    TUMBLE(TABLE order_events, DESCRIPTOR(event_time), INTERVAL '5' MINUTES)
)
GROUP BY window_start, window_end, item_id;

Defines a Kafka-backed table with an event-time attribute and watermark, then runs a tumbling-window aggregation. The result updates continuously as events arrive.

Notice how closely it matches the Spark version: a source, an event-time column with a watermark, a 5-minute tumbling window and a sum. The concepts are the same in every engine. Learn them once and you can read any streaming job.

A practical rule: if you already run Spark or Databricks and a few seconds of latency is fine, use Structured Streaming. If you need very low latency, complex state or event-driven logic, look at Flink. Other options include Kafka Streams (a Java library that runs inside your own application and reads and writes Kafka), ksqlDB, and the streaming features of cloud warehouses and platforms (for example Snowflake and BigQuery). Check each product's current documentation, because streaming features change quickly.

Typical questions: event time versus processing time; what a watermark is and its trade-off; tumbling versus sliding versus session windows; how checkpoints give fault tolerance; what exactly-once means and when it holds; micro-batch versus record-at-a-time; and "design a real-time dashboard for orders" (Kafka for ingestion, a stream processor for windowed aggregates, a serving store for the dashboard, and a batch job to correct and backfill). A good answer also mentions what happens to late and duplicate events.

You are a streaming data engineer teaching a beginner. Explain this [Spark Structured Streaming / Flink SQL] job in plain English, step by step. Tell me what the event time column is, what the window and watermark do, what state the engine keeps, and what would happen to an event that arrives 15 minutes late. Then suggest one improvement.

[paste your job]

What to expect: The answer should name the window size, the watermark delay, say that late events past the watermark are dropped or handled separately, and mention checkpointing. Check claims about exact engine behaviour in the official docs.