PySpark Data Streaming

Spark offers two streaming APIs. The original Spark Streaming API is built on DStreams, sequences of RDDs produced every batch interval; it is now a legacy API and receives no new features. Structured Streaming, built on DataFrames and the Spark SQL engine, treats a stream as an unbounded table that grows with every new record and lets you query it with the same DataFrame operations used for batch jobs. New PySpark work, including everything on Databricks, should use Structured Streaming. The concepts listed in section 2 apply to both APIs; the worked example and diagrams use Structured Streaming.

1. The unbounded-table model

You write a query as if the input were a static table. Spark plans it once, then on each trigger (by default as soon as the previous micro-batch finishes) reads the new offsets from the source, runs the incremental plan over just those rows, updates any aggregation state, and writes results to the sink. Offsets and state are recorded in the checkpoint directory before results are committed, which, with a replayable source and an idempotent or transactional sink such as Delta, gives end-to-end exactly-once results.

2. Key concepts

• Real-Time Data Processing:

  PySpark Streaming enables the processing of live data streams, allowing you to handle continuous data input, like logs, sensor data, or tweets.

• DStream (Discretized Stream):

  The core abstraction in PySpark Streaming is the DStream, which represents a continuous stream of data divided into small batches (micro-batches). Each batch is a Resilient Distributed Dataset (RDD).

• Window Operations:

  PySpark Streaming allows window-based operations on DStreams. This means you can apply transformations over a sliding window of data, useful for analyzing trends or patterns within specific time frames.

• Fault Tolerance:

  PySpark Streaming is fault-tolerant. If a node fails, Spark’s RDDs ensure that the lost data can be recomputed from the original source, providing resilience against hardware or network failures.

• Integration with Batch and SQL Processing:

  PySpark Streaming integrates seamlessly with Spark's batch and SQL processing capabilities. You can use the same API for batch and stream processing, simplifying the development process.

• Stateful Transformations:

  PySpark Streaming supports stateful transformations, allowing you to maintain and update state information over time. This is useful for tracking running counts, averages, or other metrics over a period.

• Backpressure Handling:

  PySpark Streaming handles backpressure, ensuring that the system can adjust the rate at which data is ingested and processed to avoid overwhelming resources.

• Support for Various Data Sources:

  PySpark Streaming can ingest data from various sources, including Kafka, Kinesis, Flume, HDFS, and TCP sockets, providing flexibility in integrating with different data ecosystems.

• Ease of Scalability:

  PySpark Streaming is designed to scale horizontally, meaning you can add more nodes to the cluster to handle larger volumes of data without changing the application code.

• Checkpoints:

  PySpark Streaming supports checkpointing to save the state of the streaming computation periodically. This is crucial for recovering from failures and restarting the application without data loss.

• Structured Streaming:

  PySpark also supports Structured Streaming, an evolution of DStreams, which provides a more declarative and consistent API for stream processing using DataFrames and Datasets.

3. Worked example: Kafka to Delta with windows

from pyspark.sql import SparkSession, functions as F, types as T

spark = SparkSession.builder.appName("clicks-stream").getOrCreate()

schema = T.StructType([
    T.StructField("user_id", T.StringType()),
    T.StructField("page", T.StringType()),
    T.StructField("event_time", T.TimestampType()),
])

raw = (spark.readStream.format("kafka")
       .option("kafka.bootstrap.servers", "broker:9092")
       .option("subscribe", "clicks")
       .option("startingOffsets", "latest")
       .load())

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

counts = (events
          .withWatermark("event_time", "10 minutes")              # how late data may arrive
          .groupBy(F.window("event_time", "5 minutes"), "page")
          .count())

query = (counts.writeStream
         .format("delta")
         .outputMode("append")                                     # emit a window once it is final
         .option("checkpointLocation", "/chk/clicks_by_page")
         .trigger(processingTime="1 minute")
         .toTable("analytics.clicks_by_page"))

query.awaitTermination()

The Kafka source delivers bytes, so the value is cast and parsed with an explicit schema. The watermark tells Spark it may finalize a window once the maximum event time seen exceeds the window end by ten minutes; in append mode each window is written exactly once at that point, and its state is dropped. The checkpoint location must be unique per query and kept across restarts.

4. Windows and watermarks step by step

The watermark is computed as the maximum event time seen so far minus the allowed delay. In batch 2 it reaches 12:08, which is past the end of the 12:00-12:05 window, so that window's count is final and is written. By batch 3 the watermark is 12:16, closing the 12:05 and 12:10 windows. An event stamped 12:03 that arrives afterwards is older than the watermark and is dropped rather than reopening a window that has already been written. Choosing the delay is a trade-off between completeness and how long results (and state) are held back.

5. DStreams vs Structured Streaming

6. Pitfalls

• Changing a query that shares a checkpoint. Many changes (new aggregations, different keys) are incompatible with existing state; start a new checkpoint.
• Unbounded state. Aggregations or stream-stream joins without a watermark keep state forever.
• Small files. Frequent triggers write many small files; use longer triggers, Delta auto-optimize, or periodic compaction.
• Schema inference. Streaming file sources require an explicit schema unless inference is enabled; drift breaks queries.
• Output mode mismatch. Append mode with aggregations requires a watermark; complete mode rewrites the whole result each trigger.

7. Related topics

• PySpark S3 small-files compaction
• PySpark streaming and I/O
• Delta Lake - the transactional sink

Featured Videos