What is Stream Processing?

How it works

A streaming job never “finishes” its input. It reads from a replayable source such as a Kafka topic, processes records as they arrive and writes results continuously. That raises problems batch jobs do not have:

• State. Counts, windows, joins and deduplication need memory of earlier events, held in a state store that must survive restarts.
• Event time and lateness. Events arrive late and out of order, so windows are defined on event time and closed by a watermark.
• Fault tolerance. The engine records its source offsets and state in a checkpoint so it can resume where it stopped. Combined with an idempotent or transactional sink, that gives exactly-once results.
• Always-on cost and operations. Compute runs all the time, and lag must be monitored.

The main engines differ in shape. Spark Structured Streaming runs a stream as a series of micro-batches and suits pipelines whose main sink is a lakehouse table. Kafka Streams is a Java library for Kafka-to-Kafka processing inside your own service. Flink runs its own cluster and handles large stateful jobs with advanced event-time features.

Use streaming when a real requirement needs low latency (fraud scoring, inventory sync). For a dashboard refreshed every 15 minutes, micro-batch or frequent batch is usually enough.

Example

A Structured Streaming query keeps a running count per user across two arrivals of data. The second availableNow run reads only the new file, and the state from the checkpoint carries the earlier counts forward (PySpark 4.2, local mode).

import json, os, tempfile
from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder.master("local[2]").appName("glossary-streaming")
         .config("spark.sql.shuffle.partitions", "2").getOrCreate())
spark.sparkContext.setLogLevel("ERROR")
base = tempfile.mkdtemp()
os.makedirs(f"{base}/in")

def land(name, users):
    with open(f"{base}/in/{name}", "w") as fh:
        fh.write("\n".join(json.dumps({"user": u}) for u in users))

def run():
    counts = (spark.readStream.schema("user STRING").json(f"{base}/in")
              .groupBy("user").agg(F.count("*").alias("events")))
    q = (counts.writeStream.outputMode("complete").format("memory").queryName("counts")
         .option("checkpointLocation", f"{base}/chk").trigger(availableNow=True).start())
    q.awaitTermination()
    return [tuple(r) for r in spark.sql("SELECT * FROM counts ORDER BY user").collect()]

land("1.json", ["a", "b", "a"])
print("after file 1:", run())
land("2.json", ["b", "c"])
print("after file 2:", run())
spark.stop()

after file 1: [('a', 2), ('b', 1)]
after file 2: [('a', 2), ('b', 2), ('c', 1)]

Common mistakes

• Building a streaming pipeline for data nobody reads more than once a day.
• Aggregating by arrival time, which puts late events in the wrong window.
• Deleting or sharing a checkpoint directory, which loses offsets and state or mixes two queries.
• Writing to a sink that is neither idempotent nor transactional, so every restart duplicates output.