Event time vs processing time in Structured Streaming

• Event time – The timestamp recorded inside the event itself, marking when the transaction actually happened at the source (branch, ATM, mobile app).
• Processing time – The wall-clock time on the Spark executor when the event is received and processed.
• Correctness – Networks and systems introduce delays, so events arrive late and out of order; windowing by processing time misfiles delayed events, while windowing by event time always gives the correct bucket.
• Reproducibility – Replaying the same data with event-time windows always yields the same result regardless of ingestion speed; processing-time windows produce different results on every replay.
• Specification – You specify the event-time column by making it a TimestampType field in the streaming DataFrame and referencing it in withWatermark and window calls.

What is event time?

Event time is the timestamp embedded in the data itself, recording when the action actually occurred at the source. In a banking system, when Rajesh transfers money from his account at the National Bank of India, the transaction message carries a timestamp showing the exact moment the transfer was initiated, regardless of when the message reaches the streaming engine.

import org.apache.spark.sql.types.*;
import static org.apache.spark.sql.functions.*;

// Define schema with event-time column
StructType txnSchema = new StructType()
    .add("transactionId", StringType, false)
    .add("accountId", StringType, false)
    .add("amount", DecimalType.apply(15, 2), false)
    .add("eventTime", TimestampType, false)  // <-- event time
    .add("branchCode", StringType, true);

Dataset<Row> transactions = spark.readStream()
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092")
    .option("subscribe", "nbi-transactions")
    .load()
    .select(from_json(col("value").cast("string"), txnSchema).as("txn"))
    .select("txn.*");

What is processing time?

Processing time is the wall-clock time on the Spark executor when the record is actually handled. It is trivially available and requires no special column. Early streaming systems windowed by processing time because it was easy, but it produces wrong results whenever events are delayed.

// Processing-time window -- uses Spark's clock, not the event's stamp
Dataset<Row> procTimeWindow = transactions
    .groupBy(
        window(current_timestamp(), "1 minute"), // wall clock
        col("accountId")
    )
    .count();
// A transaction that happened at 10:59 but arrived at 11:05
// lands in the 11:05 window -- WRONG bucket!

Why does the distinction matter for correctness?

When events arrive late or out of order, processing-time windows bucket them by when Spark happened to see them, not when they occurred. This produces incorrect aggregates and makes results non-reproducible. Event-time windows always file each event under the correct time bucket.

// Event-time window -- correct bucketing
Dataset<Row> eventTimeWindow = transactions
    .withWatermark("eventTime", "10 minutes")
    .groupBy(
        window(col("eventTime"), "1 minute"),
        col("accountId")
    )
    .agg(sum("amount").as("totalAmount"));
// A transaction at 10:59 that arrives at 11:05
// still lands in the [10:59, 11:00) window -- CORRECT!

Event: txnId=TXN-9981, eventTime=10:59:45, arrivedAt=11:05:12
  Processing-time window: [11:05, 11:06)  -- WRONG
  Event-time window:      [10:59, 11:00)  -- CORRECT

How do you specify the event-time column?

You ensure the event-time column is a proper TimestampType in the streaming DataFrame, parse it from the source if necessary, and reference it in the withWatermark and window calls.

// Parse event time from a Kafka JSON message
Dataset<Row> parsed = spark.readStream()
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092")
    .option("subscribe", "nbi-transactions")
    .load()
    .selectExpr("CAST(value AS STRING) as json")
    .select(from_json(col("json"), txnSchema).as("txn"))
    .select("txn.*")
    .withColumn("eventTime",
        to_timestamp(col("eventTime"), "yyyy-MM-dd'T'HH:mm:ss.SSS"));

// Use it in watermark and window
Dataset<Row> windowed = parsed
    .withWatermark("eventTime", "10 minutes")
    .groupBy(window(col("eventTime"), "1 minute"), col("branchCode"))
    .agg(count("*").as("txnCount"));

How is event time different from processing time?

| Aspect | Event Time | Processing Time |
|---|---|---|
| Source | Embedded in the data | Wall clock on executor |
| Late events | Correctly bucketed by real time | Misfiled into wrong window |
| Reproducibility | Same data always gives same result | Different runs give different results |
| Requires | Timestamp column + watermark | Nothing special |
| State management | Watermark bounds state | No watermark needed |
| Use case | Financial aggregates, auditing | Simple monitoring, throughput metrics |

// Reproducibility test:
// Run 1: slow ingestion, events delayed 5 seconds
// Run 2: fast ingestion, events arrive instantly
// Event-time windows: identical results both runs
// Processing-time windows: different results each run

What role does the watermark play with event time?

The watermark makes event-time windowing bounded. Because events can be arbitrarily late, Spark could keep every window's state forever. The watermark declares how late an event may be, so Spark tracks the maximum event time seen, subtracts the watermark delay to get the boundary, drops events older than that as too late, and cleans up state for closed windows.

// Watermark of 10 minutes means:
// watermark = max(eventTime seen) - 10 minutes
// Events older than the watermark are dropped
// Windows that close before the watermark have state evicted
Dataset<Row> bounded = transactions
    .withWatermark("eventTime", "10 minutes")
    .groupBy(window(col("eventTime"), "5 minutes"), col("accountId"))
    .agg(sum("amount").as("total"));

What are edge cases?

A common edge case is events with null or corrupted timestamps. If the event-time column is null, the record is effectively dropped from event-time aggregations because it cannot be assigned to any window. You should filter or default these before the aggregation.

// Handle null event times before windowing
Dataset<Row> cleaned = transactions
    .filter(col("eventTime").isNotNull())
    .filter(col("eventTime").gt(lit("2020-01-01")))  // sanity check
    .withWatermark("eventTime", "10 minutes")
    .groupBy(window(col("eventTime"), "1 minute"), col("accountId"))
    .count();

What are common mistakes?

The most common mistake is using processing-time windows for financial aggregates where correctness matters. Another mistake is forgetting to declare a watermark when using event-time windows with Append mode, which causes Spark to reject the query.

// WRONG -- event-time aggregation in Append mode without watermark
transactions
    .groupBy(window(col("eventTime"), "1 minute"), col("accountId"))
    .count()
    .writeStream()
    .outputMode("append") // AnalysisException: watermark required
    .start();

// CORRECT -- add watermark before aggregation
transactions
    .withWatermark("eventTime", "10 minutes")
    .groupBy(window(col("eventTime"), "1 minute"), col("accountId"))
    .count()
    .writeStream()
    .outputMode("append") // Works because watermark finalizes windows
    .start();

What should a senior engineer know?

The trade-off encoded in the watermark delay is the senior-level insight. A larger delay tolerates later events and improves completeness but holds state longer and delays finalization. A smaller delay finalizes faster and uses less memory but drops more late events. In banking you tune this to the real lateness distribution of the feed, often sizing the watermark to cover the 99th-percentile delay.