How to run a Spark Structured Streaming job

Spark Structured Streaming jobs run against data that can keep arriving after the application starts, so a local proof run needs to show more than a Spark driver launch. A bounded check should start the query, feed new records, process micro-batches, expose output, and stop the query handle without leaving active streams behind.

The local run uses PySpark, a file stream source, and a memory sink in complete output mode. The script starts the stream before it atomically places JSON files into the input directory, which matches Spark's file-source expectation that new files appear as complete files rather than partial writes.

The memory sink keeps the result visible through a SQL query in the same SparkSession, while the checkpoint directory records streaming progress for the running query. Use durable sinks and shared checkpoint storage for production pipelines; the local paths are disposable proof surfaces for validating the streaming mechanics.

1. Choose the local streaming proof values.

   Master: local[2]
   Application name: sg-structured-streaming-run
   Input path: /tmp/sg-structured-streaming/input
   Checkpoint path: /tmp/sg-structured-streaming/checkpoint
   Query name: event_totals
   Sink: memory table

   The script recreates /tmp/sg-structured-streaming. Change the path before running it on a machine where that directory might contain real input files or checkpoints.

2. Save the streaming job as sg_structured_streaming_run.py.

   sg_structured_streaming_run.py

   from pathlib import Path
   import json
   import shutil
    
   from pyspark.sql import SparkSession
   from pyspark.sql import functions as F
   from pyspark.sql import types as T
    
    
   base_dir = Path("/tmp/sg-structured-streaming")
   input_dir = base_dir / "input"
   checkpoint_dir = base_dir / "checkpoint"
    
   shutil.rmtree(base_dir, ignore_errors=True)
   input_dir.mkdir(parents=True)
    
   spark = (
       SparkSession.builder
       .master("local[2]")
       .appName("sg-structured-streaming-run")
       .config("spark.ui.showConsoleProgress", "false")
       .getOrCreate()
   )
   spark.sparkContext.setLogLevel("ERROR")
    
   schema = T.StructType(
       [
           T.StructField("event", T.StringType(), False),
           T.StructField("amount", T.IntegerType(), False),
       ]
   )
    
   events = spark.readStream.schema(schema).json(str(input_dir))
   totals = events.groupBy("event").agg(F.sum("amount").alias("total"))
    
   query = (
       totals.writeStream
       .format("memory")
       .queryName("event_totals")
       .outputMode("complete")
       .option("checkpointLocation", str(checkpoint_dir))
       .start()
   )
    
    
   def add_batch(name, rows):
       tmp_file = input_dir / f".{name}.tmp"
       final_file = input_dir / name
       with tmp_file.open("w", encoding="utf-8") as handle:
           for row in rows:
               handle.write(json.dumps(row) + "\n")
       tmp_file.replace(final_file)
    
    
   def show_totals(label):
       print(label)
       spark.sql(
           "SELECT event, total FROM event_totals ORDER BY event"
       ).show(truncate=False)
    
    
   try:
       print(f"query_name={query.name}")
       print(f"query_active={query.isActive}")
    
       add_batch(
           "batch-001.json",
           [
               {"event": "checkout", "amount": 4},
               {"event": "search", "amount": 2},
               {"event": "checkout", "amount": 3},
           ],
       )
       query.processAllAvailable()
       show_totals("after_batch_1")
    
       add_batch(
           "batch-002.json",
           [
               {"event": "search", "amount": 5},
               {"event": "view", "amount": 8},
           ],
       )
       query.processAllAvailable()
       show_totals("after_batch_2")
    
       print(f"last_progress_batch={query.lastProgress['batchId']}")
       print(f"last_progress_input_rows={query.lastProgress['numInputRows']}")
   finally:
       query.stop()
       print(f"query_active_after_stop={query.isActive}")
       print(f"active_streams_after_stop={len(spark.streams.active)}")
       spark.stop()
       shutil.rmtree(base_dir, ignore_errors=True)
       print("cleanup_done=true")

   processAllAvailable() waits for Spark to process the files already present in the streaming source. The temporary filename plus replace() keeps each JSON batch from being read before it is complete.

3. Submit the streaming job in local mode.

   $ spark-submit --master local[2] sg_structured_streaming_run.py
   WARNING: Using incubator modules: jdk.incubator.vector
   Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties
   ##### snipped #####
   query_name=event_totals
   query_active=True
   after_batch_1
   +--------+-----+
   |event   |total|
   +--------+-----+
   |checkout|7    |
   |search  |2    |
   +--------+-----+
   
   after_batch_2
   +--------+-----+
   |event   |total|
   +--------+-----+
   |checkout|7    |
   |search  |7    |
   |view    |8    |
   +--------+-----+
   
   last_progress_batch=1
   last_progress_input_rows=2
   query_active_after_stop=False
   active_streams_after_stop=0
   cleanup_done=true

   Look for the second table to confirm both JSON batches were processed. query_active_after_stop=False and active_streams_after_stop=0 show that the StreamingQuery handle was stopped before the Spark session exited.

4. Remove the proof script after the run succeeds.

   $ rm sg_structured_streaming_run.py