A Beginner’s Guide to Event-Time Processing in Spark Structured Streaming#

Evolution of Streaming in Spark#

Apache Spark first introduced a streaming engine known as DStreams (Discretized Streams), which required mini-batch transformations on Resilient Distributed Datasets (RDDs). While powerful, DStreams had its own drawbacks: developers had to map from RDD transformations to streaming concepts, maintain local states in complicated ways, and manage batch-like logic for computations that were fundamentally continuous.

Spark Structured Streaming, first released in Spark 2.0, represents a leap forward. It uses the Spark SQL Engine and DataFrame-based APIs. Operations on streaming data look much like operations on static data. This unification of batch and streaming under the Spark SQL umbrella significantly simplifies a developer’s life, while also leveraging optimizations from Spark SQL’s Catalyst Engine.

Key Concepts: Micro-Batches and Continuous Processing#

Spark Structured Streaming offers two core modes of execution:

1. Micro-Batch Mode: Data arrives in small epochs, often referred to as micro-batches or mini-batches. The default approach in Structured Streaming is to treat incoming data in short time intervals, process it as a batch, and move on to the next interval.

2. Continuous Processing Mode: A more advanced (though less commonly used) option. Instead of forming micro-batches, Spark tries to process data continuously as it arrives. This reduces latency, but it comes with some restrictions and complexities in the API.

For most use cases, Micro-Batch Mode offers a balanced and straightforward approach. We’ll focus primarily on micro-batch-based examples in this blog.

Why Event-Time Matters#

Imagine a scenario where you’re collecting server logs from a distributed, globally deployed set of applications. The actual time an event (e.g., user click) occurred might precede the time your system receives it by seconds, minutes, or even hours, due to network latencies, temporary outages, or offline devices. If your aggregations and analytics rely solely on when the data arrives at the Spark engine (processing-time), you could be drawing incorrect conclusions.

Event-time processing recognizes the time embedded in each record—often a timestamp included by the generating system as “eventTime” or “timestamp.” By using this field as the reference for grouping, aggregating, and analyzing data, you reflect reality far more accurately. This is particularly important for:

• Analytics: Generating accurate dashboards of user behavior by hour, day, or any time period.
• Billing: Ensuring charges are computed strictly for the time usage was reported, even if the log arrived late.
• Real-Time Monitoring: Timely triggers or alerts based on when events actually happen, not just when they show up.

Common Challenges in Event-Time#

1. Out-of-Order Events: Data might not be received in the same order events occurred.
2. Late Data: Events can arrive minutes or hours after others that relate to the same time window.
3. Variable Latency: Some data sources have ephemeral network issues or are offline for a period, delaying the arrival of event-time data.

Spark Structured Streaming addresses these challenges via watermarks and windowing concepts that allow you to define how much lateness is acceptable before dropping or ignoring events for a specific time window.

Why Watermarks are Essential#

When your application processes streaming data, and you keep track of event-time windows, you technically need to hold onto state for all arrivals that could affect your aggregations. For instance, if you are counting the number of clicks per minute for each user, you need to keep track of all clicks that might arrive within the relevant minute. But what if data arrives a whole month late? Do you keep state for all possible durations?

Watermarking solves this problem by telling Spark how late data can arrive before it’s considered too late to influence results. For example, you might define a watermark of 10 minutes. That tells Spark: “If an event has a timestamp older than 10 minutes behind the current maximum event-time, we will no longer update previous windows, and we can safely drop that data from state.”

Configuring Watermarks#

Here’s the canonical approach to setting a watermark in Spark Structured Streaming:

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

4
val spark = SparkSession.builder
5
  .appName("EventTimeExample")
6
  .getOrCreate()
7

8
// Sample read from Kafka
9
val inputDF = spark.readStream
10
  .format("kafka")
11
  .option("kafka.bootstrap.servers", "localhost:9092")
12
  .option("subscribe", "events")
13
  .load()
14

15
// Convert binary Kafka 'value' to string, parse columns
16
val events = inputDF.selectExpr("CAST(key AS STRING) as key", "CAST(value AS STRING) as value")
17
  .withColumn("eventTime", to_timestamp(col("value").substr(1,19), "yyyy-MM-dd HH:mm:ss")) // example
18

19
val watermarked = events
20
  .withWatermark("eventTime", "10 minutes")

In the above snippet:

• withWatermark("eventTime", "10 minutes") signals to Spark that eventTime is our logical event-time column, and we allow for data that’s up to 10 minutes late relative to the newest event we’ve seen.
• After 10 minutes have passed since the maximum observed eventTime, Spark can drop any data that arrives with a timestamp older than that threshold.

This significantly reduces the memory needed to store old states for potential re-computation. Under the hood, Spark will keep track of the maximum event-time seen so far and use that for comparison.

Managing State and Memory Constraints#

State can explode if you have a large user base, frequent events, or extended periods of allowed lateness. To keep everything manageable:

1. Tune Watermarks: A shorter watermark (e.g., 5 minutes) means you free state earlier at the cost of discarding truly late arrivals.
2. Prune State: Spark automatically handles some pruning, but you can limit the scope of states (e.g., partitioning data flows).
3. Checkpoints: Store progress in a reliable checkpoint location (e.g., HDFS or cloud storage). If your driver restarts, Spark can reconstruct streaming state from these checkpoints.

Tumbling Windows#

A tumbling window has a fixed size (e.g., 5 minutes) and does not overlap with other windows. If you’re analyzing traffic data in 5-minute increments, you might have windows such as [12:00, 12:05), [12:05, 12:10), etc.

Example snippet for a tumbling window:

1
import org.apache.spark.sql.functions._
2

3
val windowedCounts = events
4
  .withWatermark("eventTime", "10 minutes")
5
  .groupBy(
6
    window(col("eventTime"), "5 minutes"),
7
    col("key")
8
  )
9
  .count()

Here, window(col("eventTime"), "5 minutes") segments data into 5-minute intervals.

Sliding Windows#

Sliding windows overlap in time. For instance, you can create a 5-minute window that slides every minute. This approach is useful when you want more fine-grained continuous analytics (e.g., a rolling average).

1
val slidingCounts = events
2
  .withWatermark("eventTime", "10 minutes")
3
  .groupBy(
4
    window(col("eventTime"), "5 minutes", "1 minute"),
5
    col("key")
6
  )
7
  .count()

In this example, each new event can appear in multiple overlapping windows.

Session Windows#

Session windows group events by user activity sessions, typically determined by a gap of inactivity. For instance, you can specify a session window of 30 minutes for a user. If no events arrive for that user within 30 minutes, the session ends, and a new one starts.

1
val sessionized = events
2
  .withWatermark("eventTime", "10 minutes")
3
  .groupBy(
4
    session_window(col("eventTime"), "30 minutes"),
5
    col("key")
6
  )
7
  .agg(count("*").as("sessionEventCount"))

Session windows handle user-based activity bursts gracefully and automatically stitch events together into sessions.

Window Aggregation Examples#

You aren’t limited to counts. Any aggregator that can handle partial and incremental updates (i.e., associative and commutative aggregates) will work, such as sum, avg, max, min, or custom aggregator expressions.

A simple example might be:

1
val aggregated = events
2
  .withWatermark("eventTime", "10 minutes")
3
  .groupBy(
4
    window(col("eventTime"), "5 minutes"),
5
    col("key")
6
  )
7
  .agg(
8
    sum("someMetric").as("totalMetric"),
9
    avg("someMetric").as("averageMetric")
10
  )

Trade-offs of Allowing Late Data#

If you allow late data to be included in older windows, your results become more accurate, but you hold onto state longer. Increasing watermark duration prevents losing important events but requires additional memory and can increase the complexity of how frequently aggregates update.

Developers must balance these concerns:

• Higher latency or memory usage if you keep windows open longer.
• Loss of correctness if you allow windows to close too soon and discard late data.

Watermark Policies for Late Data#

When you define .withWatermark("col", "duration"), Spark reclaims state only after the “col” is older than the newest event-time seen minus the specified duration. The maximum event-time observed becomes crucial. For instance, if the maximum eventTime is 12:20, and your watermark is 10 minutes, any event older than 12:10 can be dropped from state.

Re-opening Windows and Correctness#

Spark will not “re-open” windows that are considered closed by virtue of the watermark. As soon as Spark decides a window is past the threshold, new arrivals for that window are ignored. Therefore, if you do get an extremely late event for that window, it can’t change your aggregation result. This is a fundamental trade-off for robust state management in streaming systems.

Sample Dataset and Schema#

Let’s illustrate the steps with a realistic scenario. Assume you have a clickstream log with the following fields:

In streaming logs, these might come from a Kafka topic in JSON format.

Example Code in Scala and Python#

Below is a sample of how you might read this streaming clickstream data in Spark, parse the JSON, apply a watermark, and perform a windowed aggregation based on eventTime.

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

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

10
    // Read from Kafka
11
    val rawEvents = spark.readStream
12
      .format("kafka")
13
      .option("kafka.bootstrap.servers", "localhost:9092")
14
      .option("subscribe", "clickstream-topic")
15
      .load()
16

17
    // Parse JSON
18
    val events = rawEvents.selectExpr("CAST(value AS STRING) as json)
19
      .select(
20
        get_json_object(col("json"), "$.userId").as("userId"),
21
        to_timestamp(get_json_object(col("json"), "$.eventTime"), "yyyy-MM-dd HH:mm:ss").as("eventTime"),
22
        get_json_object(col("json"), "$.pageId").as("pageId"),
23
        get_json_object(col("json"), "$.referrer").as("referrer"),
24
        get_json_object(col("json"), "$.sessionId").as("sessionId"),
25
        get_json_object(col("json"), "$.eventType").as("eventType"),
26
        get_json_object(col("json"), "$.eventValue").cast("double").as("eventValue")
27
      )
28

29
    // Watermark and window aggregation
30
    val stats = events
31
      .withWatermark("eventTime", "10 minutes")
32
      .groupBy(
33
        window(col("eventTime"), "5 minutes"),
34
        col("pageId")
35
      )
36
      .agg(
37
        count("*").as("totalEvents"),
38
        avg("eventValue").as("avgValue")
39
      )
40

41
    // Start writing to console for demo, but in practice you’d write to storage or sink
42
    val query = stats.writeStream
43
      .outputMode("update")
44
      .format("console")
45
      .option("truncate", "false")
46
      .start()
47

48
    query.awaitTermination()
49
  }
50
}

1
from pyspark.sql import SparkSession
2
from pyspark.sql.functions import get_json_object, to_timestamp, col, window, count, avg
3

4
spark = SparkSession.builder \
5
    .appName("ClickstreamApp") \
6
    .getOrCreate()
7

8
# Read from Kafka
9
rawEvents = spark.readStream \
10
    .format("kafka") \
11
    .option("kafka.bootstrap.servers", "localhost:9092") \
12
    .option("subscribe", "clickstream-topic") \
13
    .load()
14

15
# Parse JSON
16
events = rawEvents.selectExpr("CAST(value AS STRING) as json") \
17
    .select(
18
        get_json_object(col("json"), "$.userId").alias("userId"),
19
        to_timestamp(get_json_object(col("json"), "$.eventTime"), "yyyy-MM-dd HH:mm:ss").alias("eventTime"),
20
        get_json_object(col("json"), "$.pageId").alias("pageId"),
21
        get_json_object(col("json"), "$.referrer").alias("referrer"),
22
        get_json_object(col("json"), "$.sessionId").alias("sessionId"),
23
        get_json_object(col("json"), "$.eventType").alias("eventType"),
24
        get_json_object(col("json"), "$.eventValue").cast("double").alias("eventValue")
25
    )
26

27
# Watermark and window
28
stats = events \
29
    .withWatermark("eventTime", "10 minutes") \
30
    .groupBy(
31
        window(col("eventTime"), "5 minutes"),
32
        col("pageId")
33
    ) \
34
    .agg(
35
        count("*").alias("totalEvents"),
36
        avg("eventValue").alias("avgValue")
37
    )
38

39
# Write to console for demonstration
40
query = stats.writeStream \
41
    .outputMode("update") \
42
    .format("console") \
43
    .option("truncate", "false") \
44
    .start()
45

46
query.awaitTermination()

Trigger Options#

Triggers determine how frequently Spark checks for new data or finalizes micro-batches. Common triggers:

1. ProcessingTime(“10 seconds”): Spark will generate a batch every 10 seconds if data is available.
2. Once: Process available data once, then stop (often used in incremental batch contexts).
3. Continuous(“1 second”): For continuous processing mode with sub-second latencies. (More advanced, less widely used.)

Example usage:

1
val query = stats.writeStream
2
  .format("console")
3
  .trigger(Trigger.ProcessingTime("10 seconds"))
4
  .start()

State Store Management#

Under the hood, Spark keeps state in memory and writes periodic checkpoints:

• Checkpoints: Configured via .option("checkpointLocation", "path/to/checkpoints").
• Cleanup: Spark will automatically clear state that’s no longer needed due to watermarks and closed windows. Ensure you have adequate storage space and a retention policy for checkpoint directories.

Resource Provisioning#

For large data volumes:

1. Scale horizontally: Increase the number of executors and the memory per executor.
2. Use cluster managers: Yarn, Kubernetes, or Mesos can dynamically allocate resources.
3. Optimize your cluster: Tune parallelism, data partitioning, and keep an eye on shuffle overhead.

Continuous Processing Mode#

Spark’s continuous processing mode eliminates micro-batches for ultra-low latency streaming. However, it constrains the type of operations you can perform. Event-time windows in continuous mode are somewhat constrained, and watermarks work differently. It requires more careful code and an understanding of how near-real-time data arrivals are handled.

Use continuous mode if sub-second latency is critical and your transformations are relatively simple. Otherwise, micro-batch mode with short triggers is often simpler and more widely tested.

Fault Tolerance and Checkpointing#

Spark Structured Streaming ensures once-and-only-once fault tolerance if:

• You configure a reliable checkpoint directory.
• You use idempotent sinks or output modes that can handle reprocessing upon restart.
• You handle deduplication with unique event IDs if replays from the source are possible.

Handling Large-Scale Events#

Event-time processing can lead to wide windows with large states. For instance, analyzing a 24-hour window with 2 million events per minute is vast. Some strategies to handle large scale:

1. Partition your streaming job by key, user, or otherwise chunk data to multiple parallel streaming queries, depending on the business logic.
2. Watermark aggressively if possible.
3. Use specialized sinks that can handle incremental updates more efficiently, like Delta Lake or some high-throughput store.

Integration with Kafka and Other Sources#

Spark Structured Streaming integrates seamlessly with Apache Kafka. You can also connect to file-based sources (e.g., S3, HDFS), socket streams, or big data ingestion services (e.g., AWS Kinesis). Each source has slightly different configurations, but the watermark concept remains consistent—Spark always respects your event-time field for windowing and state management.

For Kafka, special consideration is needed for:

• Offset Management: By default, Spark handles Kafka offsets in the checkpoint directory.
• Schema Evolution: When data can evolve over time, consider using schema registries (e.g., Confluent).

Data Lakes and Lakehouses#

Modern architectures commonly place streaming data into lakehouse platforms (e.g., Delta Lake on Databricks, Apache Hudi, or Apache Iceberg). These architectures enable both real-time data ingestion and historical batch analytics on the same unified data store. For event-time processing:

1. Use Merge-on-Read: Some table formats let you continuously ingest data while still providing near-real-time queries.
2. Partition by Event-Time: If your data is time-series in nature, partition your storage tables by date or hour for faster queries.

Monitoring and Alerting#

A professional-grade streaming pipeline demands continuous monitoring of:

1. Throughput: How many records per second are processed?
2. Latency: How long does it take from an event’s occurrence to its appearance in your final sink?
3. State Store Size: Watch memory usage or on-disk footprints for the streaming state.
4. Watermark Delays: Track the difference between processing-time and event-time.

You can use Spark’s built-in web UI, log-based alerts, or external monitoring platforms like Grafana to keep track of these metrics. Setting up alerts when processing falls behind or state store size grows unexpectedly can prevent disruptions.