Legacy DStreams versus Structured Streaming?

Summary

• The Spark docs call Spark Streaming (DStreams) the previous generation of Spark's streaming engine: "There are no longer updates to Spark Streaming and it's a legacy project." They tell you to use Structured Streaming for streaming applications and pipelines.
• A DStream is represented internally as a sequence of RDDs. Structured Streaming treats a stream as an unbounded table that you query with the Dataset and DataFrame APIs.
• Structured Streaming supports event-time windows and, since Spark 2.1, watermarks. It can ensure end-to-end exactly-once semantics with replayable sources and idempotent sinks.
• In DStreams, transformations are processed exactly once, but output operations are at-least-once by default.
• Migration is a rewrite of the entry point, source, transformations and sink. The Spark docs show a word count program in each API.

Legacy DStreams versus Structured Streaming?
Apache Spark has two streaming APIs. DStreams (Spark Streaming) is the older one, and the Spark docs describe it as "the previous generation of Spark's streaming engine." Structured Streaming is the engine the docs recommend for new work and for existing pipelines.

How do the two APIs differ?

Dimension	DStreams	Structured Streaming
Status in the Spark docs	Legacy project, no further updates	The engine to use for streaming applications and pipelines
Abstraction	A DStream, a sequence of RDDs	An unbounded table, queried with the Dataset and DataFrame API
Languages	Scala, Java, Python	Scala, Java, Python, R
Windows	Window length and slide interval must be multiples of the batch interval	Event-time windows on a timestamp column
Late data	The DStreams guide describes no watermark API	withWatermark sets a late-data threshold and lets the engine clean up old state
Exactly-once	Transformations exactly once, output operations at-least-once by default	End-to-end exactly-once with replayable sources and idempotent sinks
SQL and DataFrames	Convert each RDD to a DataFrame inside foreachRDD	Native, on the Spark SQL engine
Spark UI	Streaming (DStreams) tab	Structured Streaming tab

What does the same job look like in each API?

The Spark docs use a streaming word count over a TCP socket. This is the DStreams version:

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, 1)
lines = ssc.socketTextStream("localhost", 9999)
words = lines.flatMap(lambda line: line.split(" "))
pairs = words.map(lambda word: (word, 1))
wordCounts = pairs.reduceByKey(lambda x, y: x + y)
wordCounts.pprint()
ssc.start()
ssc.awaitTermination()

This is the Structured Streaming version:

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split

spark = SparkSession.builder.appName("StructuredNetworkWordCount").getOrCreate()
lines = (spark.readStream.format("socket")
    .option("host", "localhost").option("port", 9999).load())
words = lines.select(explode(split(lines.value, " ")).alias("word"))
wordCounts = words.groupBy("word").count()
query = wordCounts.writeStream.outputMode("complete").format("console").start()
query.awaitTermination()

The docs say the socket source is for testing only because it does not provide end-to-end fault-tolerance guarantees.

How do I migrate?

Map each DStreams call to its Structured Streaming counterpart:

DStreams	Structured Streaming
StreamingContext(sc, 1) and its batch interval	SparkSession, plus a trigger such as .trigger(processingTime='2 seconds')
ssc.socketTextStream(host, port)	spark.readStream.format("socket")
flatMap, map, reduceByKey	select, explode, split, groupBy, count
pprint()	writeStream.format("console")
ssc.start(), ssc.awaitTermination()	start(), query.awaitTermination()
streamingContext.checkpoint(directory)	.option("checkpointLocation", path)
reduceByKeyAndWindow	groupBy(window(...)) after withWatermark
foreachRDD	foreachBatch

How do watermarks handle late data?

Event time is the time embedded in the data itself. You name the event-time column and a threshold:

from pyspark.sql.functions import window

# words: streaming DataFrame with columns timestamp and word
windowedCounts = (words
    .withWatermark("timestamp", "10 minutes")
    .groupBy(window(words.timestamp, "10 minutes", "5 minutes"), words.word)
    .count())

The guarantee runs in one direction. Data that arrives within the delay is guaranteed to be aggregated, but data delayed by more than the threshold "is not guaranteed to be dropped; it may or may not get aggregated."

What changes on Databricks?

Databricks compute does not report activity from DStreams, so an auto-terminating compute "may be terminated while it is running DStreams." The docs advise you to turn off auto termination for compute running DStreams or to consider Structured Streaming. Databricks describes Structured Streaming as a near real-time processing engine with end-to-end fault tolerance and exactly-once processing guarantees using familiar Spark APIs.

FAQs

Is DStreams still supported?

The Spark docs call it a legacy project with no further updates, and the PySpark API reference labels the module "Spark Streaming (Legacy)." They tell you to use Structured Streaming instead.

How does Structured Streaming get exactly-once results?

Every source is assumed to have offsets, and the engine uses checkpointing and write-ahead logs to record the offset range of each trigger. Sinks are designed to be idempotent. foreachBatch gives only at-least-once writes by default, so use its batch id to deduplicate output.

Can I run SQL on a DStream?

Yes. Convert each RDD to a DataFrame inside foreachRDD, register it as a temporary view, and query it with spark.sql.