Forum Discussion
Detecting corrupted data in under 1ms: what changes when Spark stops waiting for the micro-batch
Streaming anomaly detection almost always runs into the same wall: the micro-batch. Traditional Structured Streaming processes data in one-to-two-second windows, which is fast enough for most use cases, but terrible when the very act of waiting for the batch to close is already too late. Fraud, sensitive payload validation, corrupted data entering a critical pipeline — these scenarios can't tolerate even a second of structural delay. Apache Spark's Real-time mode exists to attack exactly that gap, and Databricks published an experiment that uses a concrete use case to prove the point: analyzing Ethereum transactions in real time.
What structurally changes in Real-time mode
Unlike the micro-batch trigger, Real-time mode processes data as it arrives, without waiting for a window to close. That's possible through three combined architectural changes: a continuous flow of data between stages (instead of materializing an intermediate result on every batch), scheduling all stages of the query simultaneously (instead of scheduling stage by stage), and an in-memory streaming shuffle between stages, eliminating the cost of writing and reading shuffle files on every round.
In practice, this brings Spark Structured Streaming closer to a continuous processing model, something historically associated with engines like Flink, but within the same runtime that already runs the rest of your organization's batch and streaming pipeline.
The experiment: Ethereum transactions as a stress test
The Databricks team built a pipeline that ingests blocks and transactions from the Ethereum network and classifies each event as ALLOW or QUARANTINE, based on two checks:
- Protocol invariant validation: a block where gas_used is greater than gas_limit is logically impossible under the Ethereum protocol's own rules, so it becomes an automatic signal of corrupted or malformed data.
- Payload hygiene: the block's extra_data field is scanned for patterns that shouldn't be there, such as a fragment of PII, a JWT token, or an AWS access key leaked by accident.
The cluster used was Databricks Runtime 16.4 LTS, four i3.xlarge workers, a dedicated (single-user) cluster, with Photon deliberately disabled to isolate the effect of Real-time mode itself.
The numbers behind the promise
- Sustained input rate of approximately 65,592 rows per second.
- Sustained processing rate of approximately 69,713 rows per second (the engine was able to absorb the input without building up backlog).
- More than 23.2 million messages processed over the course of the experiment.
- P95 latency under 0.5 millisecond.
- P99 latency of 1 millisecond.
For context: a traditional 1-2 second micro-batch represents a three-order-of-magnitude difference in detection latency. If the goal is to stop a suspicious transaction before it propagates, that difference isn't cosmetic.
Hands-on: a skeleton for real-time classification
The "classify and route" pattern described in the experiment can be reproduced on any structured event schema. Here's a simplified example using transformWithState, the operator that gives you access to custom stateful logic inside Structured Streaming:
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, StringType, LongType
class BlockValidator(StatefulProcessor):
def init(self, handle: StatefulProcessorHandle):
self.handle = handle
def handleInputRows(self, key, rows, timer_values):
for row in rows:
status = "ALLOW"
if row["gas_used"] > row["gas_limit"]:
status = "QUARANTINE"
elif self._contains_sensitive_pattern(row["extra_data"]):
status = "QUARANTINE"
yield {"block_id": row["block_id"], "status": status}
def _contains_sensitive_pattern(self, payload: str) -> bool:
markers = ["AKIA", "eyJhbGciOi", "-----BEGIN"]
return any(m in (payload or "") for m in markers)
query = (
spark.readStream.table("bronze.eth_blocks")
.groupBy("block_id")
.transformWithState(
statefulProcessor=BlockValidator(),
outputStructType=StructType([
StructField("block_id", LongType()),
StructField("status", StringType()),
]),
)
.writeStream
.trigger(availableNow=False) # real-time trigger is configured at the cluster/runtime level
.toTable("silver.eth_blocks_classified")
)Worth reinforcing: enabling Real-time mode itself depends on runtime- and cluster-level configuration, it's not just a matter of swapping the trigger in the API, so the official reference documentation should be consulted before running this in production.
In practice: I'd test this kind of pipeline under real load spikes, not just a constant rate, before blindly trusting the p99 number. Real Ethereum network traffic has much more irregular transaction bursts than a synthetic constant-throughput test, and it's under bursts that any low-latency system tends to reveal its real queueing behavior.
Why Ethereum was a smart benchmark choice
It's worth pausing for a moment on why Ethereum was chosen as the test scenario, because it isn't obvious at first glance. Ethereum transactions have two properties that make them an honest stress test for an anomaly detection system: first, the volume is public and replicable, anyone can audit the dataset used, which reduces the chance of an artificially favorable benchmark. Second, and more important, the protocol's invariant rules (like gas_used never being allowed to exceed gas_limit) are known and deterministically verifiable, so there's no ambiguity about what counts as a real anomaly versus a false positive. That removes a confounding variable common in anomaly-detection benchmarks, where the very definition of "anomaly" is already subjective. Applied to a corporate scenario, the equivalent would be having clear, auditable business rules (credit limit, expected value range, field format) before trying to apply this kind of real-time classification; without that, the technical pipeline works fine, but classification quality collapses.
What this doesn't solve
Real-time mode drastically reduces processing latency, but it doesn't replace the need for a good response strategy for suspicious data. Classifying something as QUARANTINE in one millisecond is pointless if the human or automated process that handles that alert still takes minutes or hours to act. It's also important to note the documented limitations of transformWithState in real-time mode: transformWithStateInPandas isn't supported in this mode, and state management (timers, TTL) has its own rules that differ from the traditional micro-batch mode, which requires extra care when migrating an existing pipeline.
Wrapping up
The Ethereum experiment works as a proof of concept for something broader: once Spark Structured Streaming itself can operate in the millisecond range, scenarios that used to require leaving the Spark ecosystem (fraud, critical validation, immediate operational alerting) start to fit inside the same runtime that already processes the rest of your data pipeline. It's worth testing, but with a realistic expectation of what still depends on human decision-making and response after detection.
References
- Ultra-fast anomaly detection using Apache Spark Real-Time Mode (https://www.databricks.com/blog/ultra-fast-anomaly-detection-using-apache-spark-real-time-mode) (official Databricks blog)
- Apache Spark Structured Streaming Real-Time Mode: concepts (https://docs.databricks.com/aws/en/structured-streaming/real-time/concepts) (official documentation)
- Real-time mode concepts (https://learn.microsoft.com/en-us/azure/databricks/structured-streaming/real-time/concepts) (Microsoft Learn)
- Stateful applications with transformWithState (https://docs.databricks.com/aws/en/stateful-applications/) (official documentation)
- Build a custom stateful application with transformWithState (https://learn.microsoft.com/en-us/azure/databricks/stateful-applications/) (Microsoft Learn)