Every StopFire pipeline I had built up to this point started with a file already sitting on disk. Read the CSV, partition it, join it, score it, write the result. That is batch thinking, and I had not noticed how deep the assumption ran until I tried to model what the sensor network actually does, which is send a reading every sixty seconds, forever, with no file boundary anywhere in sight.
Batch assumes all the data is already there and you are deciding what to do with it. Streaming assumes the data is never all there, and you are deciding what to do with each piece as it shows up. That is not a small difference in code. It is a different question you ask before you write anything.
Faking a sensor network with a TCP socket
Spark Streaming needs a source, and I did not have a real sensor network sitting on my desk. What I built instead was a small Python script acting as a TCP server, emitting fabricated readings once a second, close enough in shape to the real StopFire data to test against.
import socket
import json
import time
import random
HOST = "localhost"
PORT = 9999
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((HOST, PORT))
server.listen(1)
print("Waiting for Spark to connect...")
conn, addr = server.accept()
print("Connected:", addr)
stations = ["STN-0142", "STN-0198", "STN-0231"]
while True:
reading = {
"station_id": random.choice(stations),
"timestamp": time.time(),
"temperature_c": round(random.uniform(18, 42), 1),
"humidity_pct": random.randint(10, 60),
"wind_speed_kph": round(random.uniform(5, 60), 1),
}
conn.sendall((json.dumps(reading) + "\n").encode("utf-8"))
time.sleep(1)One connection, one line of JSON per second, forever. It is a deliberately dumb generator. The point was never to simulate realistic weather, it was to have something on the other end of a socket that Spark could listen to continuously, so I could stop worrying about the source and focus on what happens once data starts arriving without a defined end.
The five lines that set up everything
Spark Streaming means DStreams, not Structured Streaming. Structured Streaming exists but the documentation still calls parts of it experimental, and DStreams is what the tutorials and the production examples I could find actually used. The setup is short enough to hold in your head completely:
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
sc = SparkContext("local[2]", "StopFireStream")
ssc = StreamingContext(sc, 5)
lines = ssc.socketTextStream("localhost", 9999)Five lines, and the third one is the one that took me a debugging session to understand properly. local[2] requests two local threads, not one. Spark Streaming dedicates one thread permanently to the receiver, the piece constantly listening on the socket for new data. If you run this with local[1], the receiver claims the only thread available and no thread is ever free to process the batches it receives. The job does not crash. It just sits there, connected, silently doing nothing, and the first time I hit this I spent twenty minutes checking my socket code before realising the master string was the actual problem.
StreamingContext(sc, 5) sets the batch interval to five seconds. Every five seconds, whatever arrived on the socket since the last batch gets bundled into an RDD and handed to whatever transformations come next. That interval is not free to shrink arbitrarily. A one-second batch interval processes more frequently but pays the Spark task-scheduling overhead more often per unit of data, and if processing one batch takes longer than the interval, batches start queuing up behind each other. Five seconds was comfortable headroom for what this pipeline needed to do per batch.
Turning lines into scored readings
Each five-second batch shows up as a DStream of raw text lines, and DStream transformations look almost identical to RDD transformations because a DStream is, mechanically, a sequence of RDDs, one per batch interval.
def parse_reading(line):
return json.loads(line)
def is_high_risk(reading):
return reading["wind_speed_kph"] > 40 and reading["humidity_pct"] < 20
readings = lines.map(parse_reading)
high_risk = readings.filter(is_high_risk)
high_risk.pprint()map and filter here run once per micro-batch, against that batch’s RDD, not against the whole stream at once, because there is no “whole stream” object to run them against. There is only ever the current batch’s RDD and whatever state you have chosen to carry forward between batches, which for this first version was nothing at all. pprint() is Spark Streaming’s version of a sanity check, printing the first few elements of each batch’s RDD to the console so I could watch readings flow past in real time while the socket generator ran in a separate terminal.
Writing each batch out to MongoDB
foreachRDD is the operation that gets a DStream’s output somewhere that outlives the Spark job, by running an arbitrary function against each batch’s RDD as it comes through.
from pymongo import MongoClient
def save_to_mongo(rdd):
if rdd.isEmpty():
return
client = MongoClient("mongodb://localhost:27017/")
db = client.stopfire
db.high_risk_readings.insert_many(rdd.collect())
client.close()
high_risk.foreachRDD(save_to_mongo)
ssc.start()
ssc.awaitTermination()This version works and I want to be honest about what is wrong with it. rdd.collect() pulls every record in the batch back to the driver before inserting, which is fine for a handful of high-risk readings every five seconds and a real problem the moment batch volume grows past what one driver process comfortably holds in memory. Opening a fresh MongoClient inside save_to_mongo on every single batch is also wasteful, connection setup has a real cost, and the better version of this opens one client per executor using foreachPartition instead of collecting everything to the driver. I did not fix either of those for this first pass. I wanted the pipeline to work end to end before I optimised the part of it that was actually fine at this data volume, and both of those choices are on my list to revisit once real sensor throughput replaces the one-reading-per-second generator.
ssc.start() and ssc.awaitTermination() are the last two lines, and they are what turns the whole thing from a description of a pipeline into a running one. Nothing before start() actually processes data. The DStream transformations chained above it are a lazy graph, exactly like RDD transformations are lazy, and Spark only begins pulling data through that graph once the streaming context starts.
Micro-batches are not the same as continuous processing
The mental model that eventually clicked: DStreams do not process events one at a time as they arrive. They accumulate whatever arrived during a fixed window, package that window as one RDD, and run the same batch-processing machinery Spark already has against it. A five-second window is not “streaming” in the sense of reacting to each event the instant it lands. It is batch processing running on a five-second clock instead of a human deciding when to press run.
That distinction mattered for the StopFire alerting case specifically. A wind gust that spikes for two seconds and settles could land entirely inside one batch window, or straddle the boundary between two windows and get split across both, with neither individual batch showing the full spike. True event-time processing (which DStreams is not) would not have this seam. For a fire-risk alert, five seconds of smoothing across a spike boundary was a trade-off I was willing to make. For something needing tighter latency guarantees, that same seam would be a genuine defect, not a rounding error.
Closing the loop with a query client
The pipeline is not finished until something reads what got written. A small polling script closed the loop:
from pymongo import MongoClient
import time
client = MongoClient("mongodb://localhost:27017/")
db = client.stopfire
while True:
latest = db.high_risk_readings.find().sort("timestamp", -1).limit(5)
for doc in latest:
print(doc["station_id"], doc["wind_speed_kph"], doc["humidity_pct"])
time.sleep(5)Nothing sophisticated, a find().sort().limit() on the same five-second cadence as the batch interval upstream, printed to a terminal instead of driving a real alert. But watching a reading appear in the generator’s terminal, then in Spark’s pprint() output a moment later, then in this query client’s output a few seconds after that, was the first time the whole shape of a continuous pipeline felt real to me rather than theoretical. Three separate processes, three separate terminals, and a five-second heartbeat connecting all of them without anyone pressing a button in between.