MAPD-B · Final Project — distributed pipeline + benchmark study
A real-time pipeline for the QUAX experiment: radio-frequency physics data streamed from S3 through Kafka to a live spectrum — processed on Dask, on Spark, and with no engine at all. Then two benchmarks to measure what the engine is actually worth.
AVERAGED POWER SPECTRUM · 2048 BINS · ONE TRACE PER BATCH — WHAT THE LIVE DASHBOARD DRAWS
01 · the problem
QUAX — the QUest for AXions — hunts dark-matter axions by
listening for a faint microwave tone in a resonant cavity. Its readout is a continuous stream of
I/Q radio samples. In this project, 31 pairs of ~64 MB binary files
(duck_i_*.dat / duck_q_*.dat) in a CloudVeneto S3
bucket stand in for the live data-acquisition stream.
Build it three ways — Dask, Spark, no engine — and measure what distribution actually buys.
02 · infrastructure
Ubuntu 24.04 · 2 vCPU · 4 GB. Runs Kafka (+ ZooKeeper), the processor driver, and the Streamlit dashboard — and doubles as a compute node.
Ubuntu 24.04 · 2 vCPU · 4 GB. Pure compute: a Dask worker or a Spark executor, depending on the engine under test.
Ubuntu 24.04 · 2 vCPU · 4 GB. Same role as worker1 — both engines get the identical three machines, symmetrically.
~/.ssh/config entry tunnels laptop → gate.cloudveneto.it → VM in a single hop03 · architecture
The producer announces filenames, not payloads. Workers pull the 64 MB directly from S3 — Kafka never becomes a network bottleneck.
If data arrives faster than compute drains it, nothing crashes — the backlog simply waits in quax_raw. That buffer is what Benchmark B later leans on.
Every processed spectrum lands in quax_processed; the dashboard is one more consumer, redrawing its plot per batch over an SSH port-forward.
04 · the physics kernel
# abridged — identical in all 7 scripts def process_physics_data(work_order): s3 = boto3.client("s3", endpoint_url=S3_URL, ...) obj_i = s3.get_object(Bucket="quax", Key=work_order["i_file"]) obj_q = s3.get_object(Bucket="quax", Key=work_order["q_file"]) data_i = np.frombuffer(obj_i["Body"].read(), dtype="<f4") data_q = np.frombuffer(obj_q["Body"].read(), dtype="<f4") signal = data_i + 1j * data_q # complex IQ stream windows = signal.reshape(-1, 2048) # rows of 2048 samples spectra = np.fft.fftshift(np.fft.fft(windows, axis=1), axes=1) power = np.abs(spectra) ** 2 return {"batch_id": work_order["batch_id"], "average": power.mean(axis=0), # the dashboard plots this "std": power.std(axis=0)}
05 · engines
result = process_physics_data(work_order)
A plain function call, in-process on the master node. No cluster, no scheduler, no serialization — the control group every engine has to beat.
cluster = SSHCluster(["master", "worker1", "worker2"], worker_options={"env": s3_keys}) future = client.submit(process_physics_data, wo)
Master spawns workers over SSH; client.submit returns a future instantly, so the live processor keeps consuming Kafka while batches compute. S3 keys ride in via worker env.
spark = (SparkSession.builder
.master("spark://master:7077") ... )
result = (sc.parallelize([wo], numSlices=1)
.map(process_physics_data).collect()[0])
A Standalone cluster — master daemon + one executor per worker — brought up for each benchmark. Keys via spark.executorEnv; executor Python pinned to the shared venv.
Same three machines, same S3 fetch, same FFT. The only variable left is the thing that schedules the work.
06 · benchmark design
Submit a batch, block until its result, submit the next. Only one task is ever in flight — which isolates exactly what an engine adds per task: cold start, steady-state time, jitter, CPU and memory.
5 independent trials × 20 batches per engine.
By construction, this test cannot show a speedup — nothing runs concurrently.
Fire 100 batches as fast as Kafka accepts them — no artificial delay — and let a real backlog form. Measure sustained throughput and how long each batch waits, end to end.
4 configurations, each isolating one variable.
This is the regime a distributed engine is actually for.
A benchmark answers exactly the question it was built to ask — so we built two.
07 · method
Each engine ran 5 independent trials; results report the mean and spread across trials. One unlucky batch — a GC pause, a slow S3 fetch — can dominate a single 20-sample stdev; it cannot manufacture a finding five times in a row.
Benchmark consumers use auto_offset_reset='latest' with no consumer group, so each trial
only sees messages sent after it is ready. run_trials.sh polls the log for
“Listening for work orders” before firing the producer.
Python fully buffers stdout when redirected to a file — without -u, the readiness line
never appears until the process exits, and every readiness poll times out. The trial still completes,
just ~60 s slower for no reason. Unbuffered fixed it.
run_with_env.sh evals only the two export S3_* lines from
~/.bashrc — credentials reach the benchmark process without ever being echoed,
printed, or committed.
Orchestration is two small shell scripts: run_trials.sh drives Benchmark A
(start engine → wait ready → fire 20 batches → wait exit → repeat ×5); run_load.sh does the
same for each Benchmark B configuration. Every batch lands as a row in a CSV under results/.
08 · benchmark A — per-task overhead
| engine | trial | cold start (s) | warm avg (s) | warm stdev (s) |
|---|
Steady state, the identical NumPy FFT dominates and the engines sit within ~0.2 s of each other. The repeatable differences live at the edges: Spark's five cold starts all land at 2.9 s or above (JVM warm-up) while Dask's never leave the 1.5–2.0 s band, and Spark's batch-to-batch jitter averages 0.33 s — roughly 80% above baseline or Dask — with its noisiest trials (0.36, 0.46 s) unmatched anywhere else.
One batch at a time, an engine is pure overhead. But that was only half the question.
09 · benchmark B — throughput under load
ThreadPoolExecutor. Isolates whether the fix is concurrency itself, or genuinely needs a cluster.
client.submit per batch as it arrives — exactly how the live processor runs, now measured.
collect() blocks, so 8 driver threads submit jobs concurrently to one shared SparkContext — a supported pattern.
load_producer.py fires 100 batches uncapped, cycling the 31 real
file pairs with fresh batch ids — real physics data reprocessed, no synthetic data.
Latency is measured from the moment Kafka accepts the work order to the moment the spectrum is done — queue time included, because queue time is the point.
10 · benchmark B — under a real backlog
end-to-end latency per batch, in arrival order · 100 batches, uncapped
| configuration | throughput | p50 | p95 | p99 |
|---|---|---|---|---|
| baseline · sequential | 0.80 /s | 64.2 s | 119.4 s | 124.2 s |
| baseline · 8 threads | 1.13 /s | 50.5 s | 84.5 s | 87.5 s |
| dask · 3 nodes | 1.20 /s | 43.5 s | 79.8 s | 82.8 s |
| spark · 3 nodes | 1.21 /s | 44.8 s | 80.2 s | 82.4 s |
With arrivals uncapped, every configuration falls behind — what differs is how fast the wait grows. Sequential processing climbs relentlessly to a 125 s worst case; both clusters bend the curve down to ~83 s, cutting median latency ~30% and lifting throughput ~50%. Dask and Spark are statistically indistinguishable from each other here.
11 · decomposing the win
sustained throughput · batches per second · 100-batch load
Going from sequential to 8 threads on the very same machine — no cluster, no engine — takes 0.80 → 1.13 batches/s. That's most of the total improvement, because each batch spends most of its ~1.4 s waiting on S3, not computing.
Adding two more machines and a real engine takes 1.13 → 1.20–1.21 batches/s. Real, repeatable — and modest for this I/O-bound workload. Heavier per-batch compute would grow this share; these VMs and this workload put the crossover about here.
Before you buy a cluster, stop doing one thing at a time.
12 · what we learned
For one batch at a time, engine choice barely moves speed, and overhead is all you buy: Spark's JVM cold start (~3.2 s vs ~1.7–2.0 s) and extra jitter are repeatable costs with no offsetting benefit at this scale. Dask stays lightest on CPU and memory.
Distribution genuinely helps: ~50% more throughput, ~30% lower median latency. But most of that is ordinary concurrency — 8 threads on one box gets you from 0.80 to 1.13 of the final 1.21 batches/s. The cluster adds the last, modest step.
A synchronous benchmark structurally cannot see scheduling wins; a load test cannot see per-task cost. Neither result is wrong — each answers only the question it was built to ask. Ask both.
Caveats: 2 vCPU / 4 GB VMs and a sub-2s, I/O-bound batch — bigger machines or heavier compute shift where
the concurrency-vs-cluster crossover sits. Known follow-up: S3 keys passed via worker env are visible in
ps aux on the worker VMs; harden via Dask's Security config or a secrets file.
Full build log, code, and raw per-batch CSVs:
github.com/amirh0ss3in/MAPDBProject