Tips and Tricks for Tuning Spark Structured Streaming Applications#

Structured vs. Classic Spark Streaming#

Classic Spark Streaming uses the DStreams (Discretized Streams) platform, which is an abstraction built on top of the RDD API. DStreams treat incoming data as a sequence of RDD micro-batches. Structured Streaming, in contrast, uses DataFrame and Dataset APIs, making it simpler to define and optimize queries.

Below is a quick reference table comparing some highlights:

Core Concept: Unbounded Table#

Spark Structured Streaming conceptualizes the input data as an ever-growing, unbounded table. Each new chunk of data is appended to this table in a streaming fashion. You define queries that operate on this virtually unbounded table. When triggers fire, Spark processes the newly available data, updates the result table, and pushes the incremental output to the specified sink.

Output Modes#

1. Append Mode: Only new rows appended to the result table since the last trigger are written to the sink. Commonly used when the query logic is a simple projection or filter and no aggregate updates are needed.
2. Complete Mode: Outputs the entire updated result table to the sink every time, usually for aggregate queries. This can be expensive.
3. Update Mode: Writes only the rows in the result table that have changed since the last trigger. Useful when you perform aggregations but only need changed results.

Selecting the correct output mode is important for balancing resource usage and ensuring accurate results. For example, if you’re working with a stateful aggregation (like a count of unique users over a window), you might choose Update Mode instead of overwriting the entire result every time.

1. Choosing the Right Trigger#

The trigger defines how often the system checks for new data and produces an output. Common trigger types:

• Fixed-Interval Micro-Batch Trigger: The most commonly used, for example .trigger(Trigger.ProcessingTime("5 seconds")). Spark will batch records received in the last 5 seconds of streaming data and process them.
• Continuous Trigger: Similar to fixed-interval micro-batch, but aims for sub-second latency by processing data continuously.
• One-Time Trigger: Used for batch jobs or for testing, effectively runs the query once and then stops.

To pick a trigger interval, consider:

• Latency Requirements: If you need near real-time results, a short interval like 1-2 seconds may be appropriate.
• Throughput Requirements: With a shorter interval, you risk overhead from frequent job scheduling. Larger intervals might help if you want to process bigger volumes at once.

Generally, you’ll experiment with a range of intervals to find the sweet spot where overhead remains low and data freshness meets your SLAs. For many real-world applications, somewhere between 2 and 10 seconds is a good starting point.

2. Batch vs. Continuous Processing#

Continuous processing was introduced in Spark 2.3 as an experimental feature. It aims to reduce micro-batch latency by scheduling tasks as soon as data arrives, rather than waiting for a trigger interval boundary. However, some transformations aren’t yet supported in continuous mode, and exactly-once guarantees might require additional considerations.

Use continuous mode when:

1. You require sub-second latency.
2. The transformations in your streaming query are supported in continuous execution.
3. You can handle potential complexities around checkpointing and state management in advanced scenarios.

For most generic use cases, you’ll likely continue to rely on micro-batch triggers with short intervals, focusing your tuning efforts on the broader Spark ecosystem configuration (e.g., memory, shuffle, concurrency).

3. Evaluating Input Sources and Formats#

Structured Streaming supports a variety of sources:

• File Sources: Parquet, JSON, CSV, etc.
• Kafka: The most common choice for high-throughput, distributed input.
• Socket: Useful for testing, but not recommended for production.
• Custom Sources: Built via Spark’s DataSource V2 API.

Each source impacts performance considerations differently. For instance, Kafka consumer concurrency is governed by the number of partitions in the topic. To maximize parallelism, ensure your Kafka topics are well-partitioned. Also, if you’re using a format like JSON, be mindful of parsing overhead. Formats like Avro or Protobuf can often be faster to decode and are more compact, reducing network I/O.

1. Memory Management and Caching#

Structured Streaming shares similar optimization principles with batch Spark jobs in terms of memory. Key areas to look at:

1. Executor Memory: If your streaming operators, such as windowing, require large amounts of data to remain in memory, you might need to increase the executor memory or offload state to disk.
2. Broadcast Joins: If you’re performing frequent joins with a small lookup table, broadcasting this smaller dataset can greatly reduce shuffle overhead.
3. Off-Heap Memory: Spark can use off-heap storage for certain operations to reduce garbage collection overhead. Configure this thoughtfully if your streaming job is sensitive to GC pauses.

1
val spark = SparkSession.builder()
2
  .appName("MyStructuredStreamApp")
3
  .config("spark.executor.memory", "4g")
4
  .config("spark.memory.offHeap.enabled", "true")
5
  .config("spark.memory.offHeap.size", "1g")
6
  .getOrCreate()

2. Stateful Streaming and State Store Tuning#

Many streaming applications perform aggregations or mapGroupsWithState operations to track data over time. This can become expensive, as Spark will manage state across micro-batches, storing and periodically checkpointing it. Key techniques:

1. Use the Right Key: Minimizing cardinality of the groupBy key reduces state size. For example, grouping by a high-cardinality column (like user ID) can explode the state store.
2. Set a Timeout: If you’re using mapGroupsWithState, specify timeouts (e.g., event time or processing time) so that Spark can prune stale keys.
3. Tune State Store Compression: Spark can compress state data. Evaluate trade-offs between CPU overhead and memory/disk usage.
4. Manage State Store Directory: Keep an eye on your state store directory size. If it grows uncontrollably, you risk running out of disk space or suffering from longer checkpoint load times.

1
import org.apache.spark.sql.SparkSession
2
import org.apache.spark.sql.functions._
3
import org.apache.spark.sql.streaming.{OutputMode, Trigger}
4

5
val spark = SparkSession.builder().getOrCreate()
6

7
val eventsDF = spark.readStream
8
  .format("kafka")
9
  .option("kafka.bootstrap.servers", "localhost:9092")
10
  .option("subscribe", "events_topic")
11
  .load()
12
  .selectExpr("CAST(value AS STRING)")
13
  .withColumn("eventTime", current_timestamp())
14

15
val aggregated = eventsDF
16
  .withWatermark("eventTime", "10 minutes")
17
  .groupBy(window($"eventTime", "5 minutes"), $"value")
18
  .count()
19

20
val query = aggregated.writeStream
21
  .outputMode(OutputMode.Update)
22
  .format("console")
23
  .option("checkpointLocation", "/tmp/checkpoints/myStructuredStateApp")
24
  .trigger(Trigger.ProcessingTime("5 seconds"))
25
  .start()
26

27
query.awaitTermination()

3. Working with Watermarks and Late Data#

Watermarks in Structured Streaming allow you to handle data that arrives late but still needs to be included in your aggregations up to a certain threshold. Spark retains state for keys that fall within the watermark criteria. This prevents unbounded growth of the state store. Proper watermark configuration ensures that:

• You don’t hold onto state for much longer than necessary.
• Late data is processed if it comes within the allowable time.

For example, if you set a watermark of 10 minutes on an event-time column, Spark can drop state for windows older than 10 minutes beyond the last seen event-time. If extreme lateness is common in your data, you may set a higher watermark, but be aware that memory usage will go up as you manage more state.

4. Checkpointing and Fault Tolerance#

A checkpoint directory is critical for fault tolerance in Structured Streaming. Spark uses this directory to store offsets, metadata about the streaming plan, and often, the state store for stateful aggregations. Recommendations:

1. Use a Reliable Filesystem: Such as HDFS or a cloud-based offering (S3, Azure Data Lake, etc.).
2. Isolate Checkpoints: Don’t mix checkpoint directories among multiple streaming queries in the same folder. Each query should have its own checkpoint location.
3. Monitor Disk Usage: Large stateful applications can accumulate huge checkpoint directories. Periodically clean them if the data is no longer needed, or configure retention policies.

1. Optimizing Shuffle Operations#

Shuffles can be a major bottleneck in any Spark application, including streaming:

• Reduce Data Skew: If certain keys are more frequent than others, that partition can grow much larger and slow down processing. Techniques like salting or using partition columns that have more uniform distribution can help.
• Increase Shuffle Partitions: By default, spark.sql.shuffle.partitions is often set to a conservative default (e.g., 200). For large clusters, you may need more partitions to distribute the workload, but be mindful that too many small tasks can also degrade performance.
• Shuffle Service: In YARN or Kubernetes deployments, ensure external shuffle services are configured and sized adequately.

2. Recovery Optimization and Spark UI Insights#

When your stream restarts, Spark recovers the last processed offsets from the checkpoint. If the state store is large, loading it into memory can take time. To minimize recovery delays:

• Use Faster Storage: A high-throughput, low-latency filesystem or SSD-based storage for state checkpointing can cut down on recovery times.
• Leverage the Spark UI: The Spark UI and History Server provide insights into shuffle read/write sizes, task failures, and stage-level parallelism. Investigate stages that take excessively long or tasks that fail frequently; performance issues in streaming are often correlated with large or skewed partitions, memory overflows, and GC bottlenecks.

3. Data Skew and Partitioning Techniques#

Data skew is a common problem in streaming workloads, especially when certain keys (e.g., popular user IDs) dominate the input stream. Symptoms of data skew:

• Some tasks take much longer to complete.
• Frequent OOM errors in the executor logs.
• High write/output times for specific partitions.

4. Cluster Sizing and Resource Allocation#

There’s no one-size-fits-all formula for cluster sizing, but these guidelines help:

1. CPU: Each streaming micro-batch must read, transform, and write data in near real-time. Having enough cores to parallelize tasks effectively is crucial.
2. Memory: If your application involves large stateful aggregations, ensure plenty of memory headroom. Watch for frequent GC cycles.
3. Executors vs. Cores: Sometimes it’s better to have more executors with fewer cores each (to reduce overhead from single-task stragglers). Experiment with your workload.
4. Auto Scaling: If you’re running in a cloud environment, you might leverage auto-scaling to handle fluctuating input loads.

1. Sample WordCount Streaming Application#

Below is a simple streaming word count example using socket source, primarily for illustration:

1
import org.apache.spark.sql.SparkSession
2
import org.apache.spark.sql.functions._
3
import org.apache.spark.sql.streaming.Trigger
4

5
object StreamingWordCount {
6
  def main(args: Array[String]): Unit = {
7
    val spark = SparkSession.builder()
8
      .appName("StreamingWordCount")
9
      .getOrCreate()
10

11
    val lines = spark.readStream
12
      .format("socket")
13
      .option("host", "localhost")
14
      .option("port", 9999)
15
      .load()
16

17
    val words = lines.as[String].flatMap(_.split(" "))
18
    val wordCounts = words.groupBy("value").count()
19

20
    val query = wordCounts.writeStream
21
      .outputMode("update")
22
      .format("console")
23
      .option("checkpointLocation", "/tmp/checkpoints/wordcount")
24
      .trigger(Trigger.ProcessingTime("5 seconds"))
25
      .start()
26

27
    query.awaitTermination()
28
  }
29
}

2. Tuning Parameters Cheat Sheet#

Below is a table summarizing common Spark Structured Streaming parameters and their typical impact:

Use these parameters wisely, always measuring performance and resource utilization. Keep in mind that each Spark environment is unique, so test systematically rather than blindly applying recommended values.