What is Structured Streaming and how it differs from DStreams

• Unbounded table model – Every incoming record is a new row appended to a table that never closes, and your query runs incrementally over only the new rows each micro-batch.
• DataFrame and SQL API – You write the same groupBy, filter, join, and select you use on static DataFrames, and Spark turns them into incremental streaming computations.
• Catalyst optimization – Every streaming query goes through predicate pushdown, column pruning, and whole-stage code generation, unlike the raw-RDD DStream path.
• Event-time and watermarks – First-class support for withWatermark and window functions so late and out-of-order events land in the correct time bucket.
• Exactly-once end-to-end – Checkpointed offsets plus idempotent or transactional sinks guarantee each record is reflected exactly once in the output.
• DStream replacement – DStreams were batch-of-RDDs with no Catalyst, no event-time, and only at-least-once; Structured Streaming is the modern successor.

What is Structured Streaming?

Structured Streaming is Apache Spark's high-level stream processing engine built on top of the Spark SQL engine. It models a live data stream as an unbounded input table that keeps growing as new records arrive, and it runs your query incrementally so each micro-batch processes only the newly arrived rows while Spark maintains the result as if the entire table had been queried from scratch.

In a banking context, Rajesh at National Bank of India writes a simple DataFrame query to count transactions per account, and Spark turns it into a continuously updating computation.

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.*;

SparkSession spark = SparkSession.builder()
    .appName("NBI-TransactionCounter")
    .getOrCreate();

Dataset<Row> transactions = spark.readStream()
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092")
    .option("subscribe", "nbi-transactions")
    .load()
    .selectExpr("CAST(value AS STRING) as json")
    .select(from_json(col("json"), transactionSchema).as("txn"))
    .select("txn.*");

Dataset<Row> counts = transactions
    .groupBy("accountId")
    .count();

counts.writeStream()
    .outputMode("update")
    .format("console")
    .option("checkpointLocation", "s3a://nbi-checkpoints/txn-count")
    .start()
    .awaitTermination();

+------------+-----+
|   accountId|count|
+------------+-----+
|ACC-10045231|   47|
|ACC-10098712|   12|
+------------+-----+

Why does Structured Streaming exist?

The legacy DStream API was built on low-level RDDs and exposed a batch-of-RDDs model where the developer manually managed time and state. It had no built-in event-time support, was not optimized by the Catalyst query planner, and could only reach at-least-once delivery semantics. Structured Streaming was created to give streaming the same declarative, optimized experience that batch queries enjoy.

// DStream legacy approach (deprecated) -- manual RDD handling
JavaDStream<String> lines = jssc.socketTextStream("host", 9999);
JavaDStream<Long> counts = lines.countByWindow(
    Durations.minutes(1), Durations.seconds(10));
counts.print();
// No Catalyst optimization, no event-time, no exactly-once

How does the unbounded table model work internally?

Spark periodically checks the source for new offsets, forms a small batch of the new rows, runs the incremental query plan, writes the output to the sink, and repeats. The micro-batch scheduler coordinates this loop. Each batch reads a range of offsets, applies the optimized physical plan, and commits.

// The micro-batch loop is internal, but you control the trigger
counts.writeStream()
    .trigger(Trigger.ProcessingTime("10 seconds")) // batch every 10s
    .outputMode("update")
    .format("console")
    .option("checkpointLocation", "s3a://nbi-checkpoints/txn-count")
    .start();

Micro-batch 0: offsets [0, 1042]   -> process 1042 rows
Micro-batch 1: offsets [1043, 2150] -> process 1108 rows
Micro-batch 2: offsets [2151, 3200] -> process 1050 rows

How does Catalyst optimize streaming queries?

Because Structured Streaming is built on Spark SQL, every streaming query goes through the Catalyst Optimizer. Predicate pushdown moves filters into the source scan, column pruning reads only the columns the query uses, and whole-stage code generation fuses operators into tight JVM bytecode. DStreams had none of this because they operated on raw RDDs.

// Catalyst pushes the filter into the Kafka read
Dataset<Row> highValue = transactions
    .filter(col("amount").gt(new java.math.BigDecimal("500000.00")))
    .select("accountId", "amount", "eventTime");
// Only 'accountId', 'amount', 'eventTime' columns are materialized
// Filter is applied at scan level, not after full deserialization

How do you use event-time and watermarks?

Structured Streaming provides first-class event-time support. You specify which column holds the real event time and attach a watermark with withWatermark. Late and out-of-order events are bucketed by when they actually happened, not when Spark saw them.

import static org.apache.spark.sql.functions.*;

Dataset<Row> windowed = transactions
    .withWatermark("eventTime", "10 minutes")
    .groupBy(
        window(col("eventTime"), "1 minute"),
        col("accountId")
    )
    .agg(sum("amount").as("totalAmount"));

windowed.writeStream()
    .outputMode("append")
    .format("parquet")
    .option("path", "s3a://nbi-data/hourly-totals")
    .option("checkpointLocation", "s3a://nbi-checkpoints/windowed")
    .start();

How is Structured Streaming different from DStreams?

| Feature | Structured Streaming | DStream (Legacy) |
|---|---|---|
| Abstraction | DataFrame / Dataset / SQL | RDD batches |
| Optimizer | Catalyst + Tungsten | None |
| Event-time | Native with watermarks | Manual, error-prone |
| Delivery | Exactly-once (checkpoint + idempotent sink) | At-least-once |
| API style | Declarative (groupBy, filter, join) | Imperative (foreachRDD) |
| State management | Automatic state store | Manual with updateStateByKey |
| Maintenance | Actively developed | Deprecated |

// Structured Streaming -- declarative, Catalyst-optimized
transactions.groupBy("accountId").count();

// DStream equivalent -- imperative, no optimization
dstream.foreachRDD(rdd -> {
    rdd.mapToPair(t -> new Tuple2<>(t.getAccountId(), 1L))
       .reduceByKey(Long::sum)
       .foreach(pair -> System.out.println(pair));
});

What are the rules and constraints?

Not every DataFrame operation is supported in streaming. Sorting on the full stream is not allowed because it would require buffering all data. Some join types require watermarks to bound state. Output mode must match the query type: Append requires that emitted rows never change, Complete requires an aggregation, and Update requires an upsert-capable sink.

// This FAILS -- full sort is not supported on streaming
transactions.sort("amount"); // AnalysisException

// This WORKS -- sort within a window after watermark finalizes
transactions
    .withWatermark("eventTime", "10 minutes")
    .groupBy(window(col("eventTime"), "5 minutes"))
    .agg(max("amount").as("maxAmount"));

What about exactly-once delivery?

Structured Streaming achieves exactly-once through two mechanisms working together. On the source side, checkpointed offsets ensure Spark replays only uncommitted batches after a crash. On the sink side, the sink must be idempotent or transactional so that replayed writes do not create duplicates.

// Exactly-once with foreachBatch and a database upsert
counts.writeStream()
    .foreachBatch((batchDf, batchId) -> {
        // Upsert by accountId -- replay is safe
        batchDf.write()
            .format("jdbc")
            .option("url", "jdbc:postgresql://nbi-db:5432/accounts")
            .option("dbtable", "account_txn_counts")
            .option("batchsize", "1000")
            .mode("append") // with ON CONFLICT UPDATE in the table
            .save();
    })
    .option("checkpointLocation", "s3a://nbi-checkpoints/upsert")
    .start();

What about micro-batch vs continuous processing?

Micro-batch is the production standard, giving latency in the hundreds of milliseconds to seconds. There is an experimental continuous mode (Trigger.Continuous) targeting millisecond latency, but it supports only simple map-like operations and is rarely used in production.

// Micro-batch (default, production standard)
counts.writeStream()
    .trigger(Trigger.ProcessingTime("5 seconds"))
    .start();

// Continuous mode (experimental, limited operations)
counts.writeStream()
    .trigger(Trigger.Continuous("1 second"))
    .start();

What should a senior engineer know?

The mental model shift from DStreams to Structured Streaming is the key insight. In DStreams you thought in terms of a sequence of RDD batches and manually managed state and time. In Structured Streaming you think in terms of one logical query against an infinite table, and Spark manages incremental execution, state, watermarks, and recovery for you. All new streaming work in banking should use Structured Streaming. DStreams are deprecated and receive no new features.

// Senior pattern: stream-stream join with watermarks
Dataset<Row> enriched = transactions
    .withWatermark("eventTime", "10 minutes")
    .join(
        accountUpdates.withWatermark("updateTime", "10 minutes"),
        expr("accountId = updAccountId AND " +
             "eventTime >= updateTime AND " +
             "eventTime <= updateTime + interval 1 hour")
    );

What are common interview follow-up questions?

Interviewers often ask how the unbounded table model handles state growth, how watermarks interact with output modes, and why DStreams could only achieve at-least-once. They also ask about the checkpoint compatibility problem: changing stateful query logic can make an existing checkpoint unusable, forcing a state reset. Being able to explain the trade-offs shows real production experience.