Moving 400+ Spark jobs from DStreams to Structured Streaming
A naive port made our hardest job twice as slow and quietly lost data. Here is what it took to reach parity and then pull ahead: bounded batches, fewer scheduling waves, a watermark guard, two timezone bugs and one very deep query plan.
Most of the data platform I work on runs on an in-house Scala framework called the Generic Parser. You describe a pipeline in configuration (source, parsing, enrichment, business rules, sinks) and the framework runs it as a Spark job. More than 400 production ETL jobs run on it, and until this year every streaming one of them used Spark’s original streaming API, DStreams.
DStreams are the legacy API. They work on RDDs, get none of the Catalyst optimiser, have no event-time semantics, and Spark’s own docs point new work at Structured Streaming. So the plan was simple on paper: move the framework’s streaming core to Structured Streaming, keep the job configuration contract identical, and let 400 jobs upgrade by bumping a library version.
The first working port was a disaster on the job we used as the benchmark, one of the highest-volume monitoring pipelines on the platform. Median batch time went from about 41 seconds on DStreams to 95 seconds. Two-thirds of batches took longer than a minute. Traffic spikes produced 300-second batches that never caught up, and some data simply vanished.
This is what fixed it. The work was split between me and a teammate who owned the JSON parsing and batch-level throughput changes; I owned correctness, the query-plan and session issues, and observability.
1. Stop counting things you are about to process
The DStream code cached every batch and counted it so it could log a row count:
inputDF.cache()
recordCount = inputDF.count()On an RDD that was already in memory, harmless. In Structured Streaming, count() is a full Spark action over every Kafka partition in the batch, run only to produce a log line, before the real work reads the same data again. On a typical batch of 60 partitions that was 30 seconds of pure waste.
The fix was to ask a cheaper question. Whether the batch is empty needs one row, not all of them, so inputDF.rdd.isEmpty() answers it. The real count comes for free at the end, from an accumulator the parser already increments as it goes.
2. Fewer, fatter partitions
A Kafka topic with 60 partitions becomes 60 Spark tasks. On a pod with 4 executor cores, every stage runs in 15 sequential waves. With DStreams those tasks were cheap because the data was already local. In Structured Streaming each one pays for network reads and deserialisation.
Coalescing to the executor core count right after ingest collapsed 15 waves into one and removed about 20 seconds of scheduling overhead per batch. coalesce rather than repartition matters here: it merges partitions without a shuffle.
3. Put a ceiling on every batch
Without a limit, a Kafka source takes everything available at the start of a batch. After an upstream outage that is millions of records in one micro-batch, which runs for minutes, falls further behind, and makes the next batch bigger still. That is the 300-second cascade.
maxOffsetsPerTrigger caps how many offsets a single batch may take, so a backlog drains as a series of normal-sized batches instead of one enormous one. We paired it with asynchronous progress tracking, which takes the offset and commit log writes off the critical path of every batch:
spark.sql.streaming.asyncProgressTrackingEnabled=true
spark.sql.streaming.asyncProgressTrackingCheckpointIntervalMs=100004. Make JSON parsing cheap
In schema-on-read pipelines, parsing JSON and inferring its schema can be most of the CPU. My teammate built two faster parser tiers behind a flag:
- jsoniter-scala, a zero-allocation JSON reader, for a 1.5–2× speed-up over Jackson while keeping the exact numeric type promotions and corrupt-record handling of the old path.
- A single-pass parser with schema interning. In production logs nearly every record has the same shape, yet a naive parser builds new
StructTypeobjects for each one. Interning shapes per partition, so identical records share one reference, cut allocations by 30–50%. A reference-equality check (t1 eq t2) then makes schema merging constant-time for the common case.
One trap along the way: an early version widened the schema across batches. A field seen once in one batch lived forever in the cached schema, and later select(col("tags.*"), col("*")) calls failed with AMBIGUOUS_REFERENCE_TO_FIELDS. Schemas are now resolved strictly per batch, matching spark.read.json exactly.
5. Guard the watermark
Structured Streaming tracks event time with a watermark: the latest event time seen, minus an allowed delay. Anything older than the watermark is treated as late and dropped.
That makes the watermark only as trustworthy as the most wrong clock in your data. One device that reports the year 2099 moves the watermark 70 years into the future, and every legitimate event after that is “late”. Nothing errors. The data just stops arriving downstream. That was our vanishing data.
The fix is a filter before the watermark sees anything:
if (dropFutureEvents) {
df = df.filter(
s"event_timestamp IS NULL OR event_timestamp <= current_timestamp() + INTERVAL $toleranceMinutes MINUTES")
}Null timestamps pass through untouched, and the tolerance is configurable per job.
6. Two timezone bugs, 5.5 hours each
Both are the kind that only show up when you diff the new output against the old, row by row.
Offsets swallowed. Timestamps like 2026-08-04T12:00:00Z or …+05:30 were parsed with a pattern that had no offset in it. Spark ignored the suffix and read the wall-clock time as if it were already in the session timezone, IST, so events landed 5.5 hours off. Strings that end in an offset now go through to_timestamp with no pattern, which keeps the real instant:
when(tsRaw.rlike("(Z|[+-]\\d{2}:?\\d{2})$"), to_timestamp(tsRaw))Converted twice. Window boundaries were passed through from_utc_timestamp(..., "Asia/Kolkata") before being written. But window boundaries are already instants, and the session was already in IST, so the offset was applied twice: 11 hours of shift. Selecting window.start directly fixed it. I also added a debug mode that logs raw and converted window boundaries per batch, which made the parallel run much easier to verify.
7. One projection, not thirty
The framework adds enrichment columns (job metadata, source identifiers and so on) by calling withColumn in a loop. Each call wraps the plan in another Project node. With 20–30 columns, the analyser and optimiser walk a very deep tree on every micro-batch, and the driver was spending 10–20 seconds per batch just planning. This is a well-known Spark trap, tracked as SPARK-25380.
Building all the columns up front and applying them in a single select produces one Project node no matter how many columns there are. Planning time dropped to under 100 ms.
val replaced = df.columns.map(c => enrich.get(c.toLowerCase).map(v => lit(v).as(c)).getOrElse(col(s"`$c`")))
val appended = newFields.map { case (name, value) => lit(value).as(name) }
df.select(replaced ++ appended: _*)8. The session you configure is not the session that runs
With DStreams, spark.conf.set("spark.sql.caseSensitive", "true") changed the one global session everything used. In Structured Streaming, a query runs on a clone of the session it was started from, so setting config on the original afterwards does nothing to the DataFrame inside foreachBatch. Payloads with both id and ID then failed with ambiguous-reference errors.
The fix is to configure the DataFrame’s own session, df.sparkSession.conf, around the one step that needs case sensitivity.
9. Don’t lose writes on the way out
While we were in there, the Kafka sink got production defaults: acks=all, idempotent producers so retries cannot create duplicates, and lz4 compression with 128 KB batches and a 50 ms linger. Idempotence requires acks=all, so a small guard turns it off for any job that overrides acks, rather than letting the producer refuse to start.
The result
We ran the tuned version side by side with the DStream job for 22 hours, 1,289 batches:
| Metric | First port | Tuned | DStream |
|---|---|---|---|
| Median batch | 95.4 s | 43.6 s | 41.2 s |
| Batches over 60 s | 68.2% | 26.4% | 24.1% |
| Worst batch | 300+ s | 72 s | 70 s |
| Cores | 12+ | 6 | 6 |
| Records lost | millions | 0 | 0 |
On the hardest job that is parity. Across the rest of the fleet, where Catalyst and whole-stage code generation have more to work with, jobs came out about 25% ahead on throughput per core. And every job now gets event-time semantics, bounded batches and proper sink guarantees for free.
The last piece was being able to see all of this while it runs. The stock Spark UI turned out to be almost no help, which is its own story: building a Spark UI for Structured Streaming.
If you are about to do the same migration
- Diff outputs row by row against the old job for at least a full day. Most of the bugs above produced plausible-looking data.
- Set
maxOffsetsPerTriggerfrom day one. An unbounded source is fine until the first outage. - Filter future timestamps before the watermark.
- Grep for
withColumninside loops. - Remember that
foreachBatchruns on a cloned session.