You're likely diving into Spark Streaming because you want to move away from batch processing and handle data as it happens. But there is a specific trap I see almost every engineer fall into when they start aggregating streaming data. You'll write code that looks perfectly logical based on your experience with standard Spark DataFrames, and then Spark will throw a cryptic AnalysisException the moment you try to start the query.

Take a look at this snippet. I'm trying to count the number of events per user in a real-time stream of logs.

val events = spark.readStream
  .format("socket")
  .option("spark.socket.host", "localhost")
  .option("spark.socket.port", 9999)
  .load()

val counts = events
  .groupBy("userId")
  .count()

val query = counts.writeStream
  .outputMode("append")
  .format("console")
  .start()

When you run this, Spark will scream at you: AnalysisException: Append output mode not supported when current aggregation is performed on streaming data with no watermark. It feels like a contradiction. You just want to append the results to your console! Why is Spark stopping you?

The "Append Mode" Wall

Here is the deal: Spark is obsessed with correctness. When you perform an aggregation (like count()), Spark has to maintain a "state" in memory to keep track of the running totals. If you use append mode, Spark promises that it will only ever output a row once it is 100% sure that the row will never change again.

In the code above, the count for "User A" could change every single second as new events arrive. If Spark appended the current count to your sink now, and then the count changed a second later, it would have to "take back" the previous row—but append mode explicitly forbids updating or deleting previously written data. Spark doesn't know when a user is "finished" sending events, so it refuses to start the query.

Solving the State Problem with Watermarks

To fix this, we have to tell Spark how long it should wait for late data before it considers a window "closed" and safe to append. This is where withWatermark comes in. You need a timestamp column in your data to make this work.

val events = spark.readStream
  .format("json")
  .schema(logSchema) // Assume we defined a schema with 'timestamp' and 'userId'
  .load("/data/logs/")

val counts = events
  .withWatermark("timestamp", "10 minutes") // Tell Spark to ignore data older than 10 mins
  .groupBy(window($"timestamp", "10 minutes"), $"userId")
  .count()

val query = counts.writeStream
  .outputMode("append")
  .format("console")
  .option("checkpointLocation", "/tmp/checkpoint")
  .start()

By adding the watermark and grouping by a window, we've given Spark a boundary. Now, Spark knows that once the system clock passes the end of a window plus the 10-minute watermark, no more data for that window is expected. It can then safely "finalize" the count and append it to the sink.

Managing the Checkpoint Location

You noticed I added .option("checkpointLocation", "/tmp/checkpoint"). Don't skip this. In a real production pipeline, if your cluster crashes, you don't want to restart your stream from the beginning of time (or lose where you left off). The checkpoint directory stores the metadata and the state of your aggregations. If the query fails, Spark looks at the checkpoint and resumes exactly where it stopped. If you're running this on a real cluster, point this to HDFS or S3, not a local folder.

Choosing Between Complete and Append Modes

Now, you might be thinking, "Why can't I just use outputMode("complete")?" You can! complete mode tells Spark to rewrite the entire result table to the sink every time there is an update. This is great for a dashboard where you only care about the current total. However, if you're writing to a database or a file system, writing the entire state every few seconds is a performance nightmare. Use complete for small lookups and append (with watermarks) for actual data pipelines.

📋 Practical Task

Build a Real-Time Error Monitor for System Logs

Your goal is to build a streaming pipeline that monitors a directory of JSON logs and alerts you when a specific service is failing. You will need to implement a windowed aggregation that counts "ERROR" level logs per service.

Requirements:

• Read a stream of JSON files from a local directory. Each JSON object contains serviceName (String), level (String), and timestamp (Timestamp).
• Filter the stream to only include rows where level == "ERROR".
• Implement a 5-minute sliding window that updates every 1 minute.
• Set a watermark of 10 minutes to handle late-arriving log files.
• Group the data by both the window and the serviceName.
• Output the results to the console using append mode.
• Ensure you include a checkpointLocation to make the pipeline fault-tolerant.

Testing Tip: Create a few JSON files with varying timestamps—some current and some slightly in the past—and drop them into your source folder while the stream is running to see the watermark and windowing in action.