How can you process real-time sensor data and calculate the average temperature per minute using Structured Streaming with windowing in PySpark?

Read the sensor stream, parse it with an explicit schema, add a watermark on the event-time column, group by a 1-minute window (and the sensor id), and compute avg("temperature"). Write the result with a checkpoint, using append mode to emit each minute once it is final, or update mode for results that update as data arrives.

from pyspark.sql import functions as F

schema = "sensor_id STRING, temperature DOUBLE, event_time TIMESTAMP"

readings = (spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "sensor-readings")
    .option("startingOffsets", "latest")
    .load()
    .select(F.from_json(F.col("value").cast("string"), schema).alias("r"))
    .select("r.*")
    .filter(F.col("temperature").isNotNull()))

per_minute = (readings
    .withWatermark("event_time", "2 minutes")
    .groupBy(F.window("event_time", "1 minute"), "sensor_id")
    .agg(F.avg("temperature").alias("avg_temperature"),
         F.count("*").alias("readings"))
    .select("sensor_id",
            F.col("window.start").alias("minute_start"),
            F.col("window.end").alias("minute_end"),
            "avg_temperature", "readings"))

query = (per_minute.writeStream
    .outputMode("append")
    .option("checkpointLocation", "/chk/sensor_avg_1min")
    .trigger(processingTime="30 seconds")
    .toTable("iot.sensor_avg_1min"))

Why each piece is there

• Event time: windows are based on when the reading was taken (event_time), not when it arrived, so delayed messages land in the right minute.
• Watermark of 2 minutes: readings up to 2 minutes late are still counted; after that, each minute's state is finalised and removed, so memory stays bounded. Choose the delay from how late your sensors' data really arrives.
• Append mode: each window's average is written once, after the watermark passes the end of the window, so the output table only ever receives final values. That adds latency (window length plus the delay). Use update mode with a sink that supports upserts (for example foreachBatch with a Delta MERGE) if a dashboard needs provisional values sooner.
• Checkpoint: stores Kafka offsets and window state, so the query resumes after a restart without double-counting.

Variations

• Sliding windows: F.window("event_time", "5 minutes", "1 minute") gives a 5-minute average every minute.
• Session windows: F.session_window("event_time", "10 minutes") groups bursts of activity.
• Testing without Kafka: the rate source or a folder of JSON files (spark.readStream.schema(schema).json(path)) works the same way.

Production concerns

• Bad messages: from_json returns null for unparseable values; filter them out and send them to a quarantine table instead of dropping them silently.
• Units and calibration: normalise temperatures (Celsius versus Fahrenheit) before averaging.
• Large numbers of sensors: use the RocksDB state store and watch state size in the streaming query's progress.
• Alerts: a second query, or foreachBatch, can flag minutes where the average crosses a threshold.