StreamingGENERALENGINE-SPECIFIC

Stateless Stream Processing

Filter, map, transform: operators whose output for a record depends only on that record. The cheapest, most restartable, most rescalable thing a stream can do — and a much narrower category than it first appears.

What actually happens How to build it Can I trust it?

Who needs this, what one row is, and why the obvious build breaks

Every lesson starts from the consumer, because designing from the source outward is this domain's characteristic mistake.

The question

Which streaming transformations can be computed from a single record, and why does that make them so much cheaper to operate?

Who needs this

Every downstream stage in the pipeline, and the people who page for it. A stateless stage gives them a stream that is parsed, typed, filtered and shaped, so the expensive stateful work that follows sees clean records — and it gives operators a stage they can restart, redeploy and rescale without thinking about state at all.

What one row is

One event in, zero or more events out, with no reference to any other event. The grain is preserved by map, reduced by filter, and multiplied by flatMap — and that last one is where a supposedly stateless stage quietly changes what one row means downstream.

The obvious build

Put all of it in one job: parse the JSON, drop the test traffic, rename the fields, look up the country from the IP, deduplicate, and aggregate into five-minute counts. It is one deployment, one set of dashboards, one thing to run.

Why it breaks

The dedup and the aggregate need state, so the whole job needs state — including the parsing stage that did not. Now redeploying a field rename requires a state migration (Streaming State).

How it breaks with real data

The dedup and the aggregate need state, so the whole job needs state — including the parsing stage that did not. Now redeploying a field rename requires a state migration (Streaming State).
The IP-to-country lookup calls an external service per record. Throughput is now bounded by that service's latency, and the job's failure modes include somebody else's outage (External Calls Inside a Transaction).
A malformed record raises inside the parse step and the operator restarts, replays the same record, and raises again. A poison message stalls the entire partition indefinitely (A Dead-Letter Queue Is a Workflow, Not a Bin).
The filter drops "test traffic" by matching an email domain. Marketing changes the domain and the filter silently stops filtering; nothing fails, and the numbers move by an amount nobody can attribute.
Rescaling for a throughput spike means rescaling the stateful part too, which means redistributing state, which means downtime that the stateless part never needed.

Source→Ingestion→Raw→Transformation→Validation→Storage Model→Serving→Consumer→Observability

What is actually happening

A stateless operator is a pure function of one record. Given the same input record it produces the same output, regardless of what came before it, what instance it runs on, or when it runs. That is the entire definition and everything convenient about it follows.
Because there is no state, any instance can process any record. The scheduler is free to move partitions between instances at will, so rescaling is a rebalance rather than a state redistribution (Consumer Groups and the Parallelism Ceiling).
Because a stateless operator is deterministic, a replay reproduces its output exactly. Recovery is "rewind and reprocess", with no snapshot to restore and no risk of restoring a stale one (Replay from the Log).
Records still have to be routed, so a stateless stage is normally *chained* into the same task as its neighbours rather than shuffled between them — filter early and the network cost of everything after it drops in proportion to what you removed.
The three shapes are filter (predicate, keeps or drops), map (one-to-one transformation of the payload) and flatMap (one-to-many, including one-to-zero). Projection — dropping fields you will not use — is a map and is the single cheapest optimisation in a streaming pipeline (Projection Pushdown).

One record in, and nothing remembered

The test for statelessness is blunt: could this operator produce its output if it were handed exactly one record and nothing else, on a machine that had just booted? If yes, it is stateless, and every convenient operational property follows automatically. If no — if it needs to know what it saw before, or what some other stream said — it is stateful, and it belongs in the next lesson with all the machinery that implies.

The reason to care is not purity. It is that this test predicts exactly how a stage behaves during the three events that dominate a streaming job's life: a redeploy, a rescale and a replay. A stateless stage survives all three trivially. A stateful one needs a plan for each.

The code below is deliberately tool-free. The point is not an API; it is that the function signature takes one record and returns a list, and that the rejection path is a return value rather than an exception. A stage that can raise can stall a partition forever on a single malformed record.

The whole of a stateless stage, with no framework in sight

1# One record in, zero or more out. No self, no cache, no clock.
2 
3def handle(raw: bytes) -> tuple[list[dict], list[dict]]:
4    """Returns (emitted, rejected). Never raises: a raise stalls the partition."""
5    try:
6        e = json.loads(raw)
7    except ValueError as err:
8        return [], [{"reason": "unparseable", "detail": str(err), "raw": raw}]
9 
10    # Reject on the fields we USE. Ignore fields we do not — an upstream
11    # addition must not become an outage here.
12    for field in ("event_id", "occurred_at", "account_id", "amount_minor"):
13        if field not in e:
14            return [], [{"reason": f"missing:{field}", "raw": raw}]
15 
16    # filter: a predicate expressed as data, not as a literal
17    if e["account_id"] in EXCLUDED_ACCOUNTS:      # loaded reference set
18        return [], []                             # dropped — and counted by the caller
19 
20    # map + projection: emit only what downstream declared it needs
21    return ([{
22        "event_id": e["event_id"],
23        "event_time": e["occurred_at"],           # the event's own clock
24        "account_id": e["account_id"],
25        "amount_minor": int(e["amount_minor"]),
26    }], [])

Three things to notice: rejection is data, not control flow; the required-field check names only the fields this stage reads; and the projection is where payload size — and therefore every downstream shuffle — is decided.

Why this is the cheap part of the pipeline

Cost in a streaming job comes from four places: bytes moved across the network, state held, work repeated after failure, and capacity held while idle. A stateless stage participates in exactly the first, which is why its cost curve is the only genuinely linear one in this module.

It is also the stage with the highest leverage over everybody else's cost. Filtering and projecting before a shuffle removes bytes from serialisation, from the network, from deserialisation and from every operator downstream. Doing the same work after the shuffle pays for all of that first and then throws the result away.

The one way to ruin this is a per-record remote call. It replaces a microsecond-scale pure function with a network round trip, so parallelism has to rise to hold throughput, and the job inherits the availability of a service it does not own. If the reference data is small enough to broadcast, broadcast it; if it is not, accept that you are building a stateful enrichment stage and design it as one.

What moves the cost of a stateless stage, relative to each other

Bytes serialised across a shuffle boundary

The dominant term, and the one a projection directly controls. Dropping unused fields before the shuffle reduces it in proportion to the share of the payload they occupied.

Per-record remote calls

Only appears if you put one here — but when it does it dominates everything, because throughput becomes a function of someone else's latency rather than of your CPU.

Deserialisation and parsing

Grows with payload size and with format verbosity; a schema-encoded binary payload costs less here than the equivalent JSON because there is less to inspect.

The transformation function itself

Almost always the smallest term and almost always the first thing people try to optimise. Regexes and date parsing are the exceptions worth profiling.

State

Zero by definition. Shown at zero deliberately: this is the only stage in the module where this row is empty, and it is the reason the stage rescales freely.

Relative weights, not measurements. The ordering is what transfers between platforms; the magnitudes do not, and this domain never publishes a price.

Relative weights for a typical parse-filter-project stage, given to establish an ordering rather than a magnitude. The teaching is the ordering: what you drop and when you drop it dominates what you compute.

The things that look stateless and are not

GENERALEvery row here is a property of the computation rather than of the engine, so the classification transfers unchanged between Flink, Spark, Kafka Streams and a hand-written consumer loop. What differs is only how loudly the engine complains: some make state explicit and declared, and a hand-written consumer makes it invisible.

Almost every incident in this category comes from a stage that was classified as stateless and was not. The classification matters because it decided the deployment strategy, the rescaling plan and whether anyone wrote a replay procedure.

The table below is the list worth memorising. Each row is a stage that passes a casual reading of "one record in, one record out" and fails the boot-a-fresh-machine test, and each has a distinct symptom that shows up long after the design decision.

The general rule underneath them: if the output depends on anything other than the bytes of this record — a cache, a clock, a counter, a previously-seen set, a sequence number — the stage is stateful and must be operated as such, no matter how the code looks.

Stages that fail the statelessness test

Trigger	Symptom	Cause	Response
Enrichment from a lookup cached in the operator	After a restart, a burst of records comes out with a missing or wrong dimension, then the problem disappears on its own.	The cache is state. A fresh instance starts cold, and records processed during warm-up get a different answer than the same records would get a minute later.	Broadcast the reference data as a stream and join against it explicitly, or accept the stage as stateful and populate the cache before processing begins (Stream Joins).
Deduplication by "have I seen this id"	Duplicates reappear after every redeploy, in a burst, then stop.	The seen-set is state and was not checkpointed, so the restarted instance believes it has seen nothing.	Make it explicitly a keyed state with a TTL, or move deduplication to the sink where it can be expressed as an idempotent upsert (Deduplication, Upserts and Merges).
A stage that uses the current time	A replay of last week produces different output than the original run did, and nobody can reproduce a number.	Wall clock is hidden state. `now()` in a streaming operator makes the output a function of when the job ran (Processing Time).	Take time from the record, not from the machine. If a processing-time decision is genuinely required, emit it as a separate field so at least it is visible (Event Time).
A sequence or surrogate key assigned in the operator	Reprocessed records get different keys than the originals, and downstream joins produce ghosts.	A counter is state, and a distributed counter is state that is also contended.	Derive keys deterministically from the record's natural key, so a replay produces the same key (Surrogate Keys).
A filter reading a mutable reference table	Two runs over the same input disagree, and the difference tracks when somebody edited the table.	The predicate depends on external mutable data — the stage is deterministic given both inputs, and only one of them is in the log.	Version the reference data and record which version produced which output range, or stream the reference table so its changes are ordered against the events (Event vs Snapshot Modeling).

How to build it

Most important first.

Separate the stateless prefix from the stateful body into different jobs, or at minimum different operators with an explicit boundary. The stateless part then gets the deployment and scaling properties it deserves instead of inheriting the stateful part's.
Filter and project as early as possible, ideally in the first operator after the source. Every byte you drop before a shuffle is a byte that is not serialised, not sent and not deserialised (The Shuffle).
Make parsing total, not partial: every record produces either a parsed record or a rejected record with the reason and the raw bytes attached. A stage that can raise is a stage that can stall a partition (A Dead-Letter Queue Is a Workflow, Not a Bin).
Resolve enrichment from a broadcast or replicated lookup rather than a per-record network call. If the reference data genuinely must be remote, cache it and accept that this stage is now stateful in every way that matters (Stream Joins).
Encode the filter's intent as data, not as a literal in code — a reference table of excluded accounts is auditable and changeable; a hard-coded domain match is a silent time bomb (Data Contracts).
Count what you drop. A filter with no counter is a data loss mechanism with no observability (Pipeline Metrics).

What this actually promises

Naming the guarantee you do not have is worth more than naming the one you do — everything downstream inherits the weakest promise in the chain.

Determinism: given the same record, the same output, always. This is the only guarantee in the entire module that holds without qualification, and it is why stateless stages are trustworthy under replay.
Restartability: no state to restore, so a restart costs only the reprocessing of uncommitted records. The output is still at-least-once, because the offset commit and the downstream write are two separate events (At-Least-Once Delivery).
Rescalability: parallelism can change up or down at any time without redistributing anything, bounded by the partition count.
What is not promised: that dropped records are recoverable (they are not, unless you route them somewhere), that ordering is preserved across a flatMap that fans out, or that a per-record external call is idempotent.

Can I trust it?

A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.

The check that would catch this

Assert on counts in versus counts out plus counts rejected per stage, and require them to balance. A stateless stage that loses records is losing them in a filter or an exception handler, and this one identity localises it immediately.
It misses a filter that is wrong rather than broken: if the predicate excludes a legitimate category, in and out still balance and the number is still wrong (Data Tests).
It also misses type coercion that succeeds and destroys meaning — a numeric string parsed as zero, a timestamp parsed with the wrong zone. Add a rejection rule that requires values to be in an expected range rather than merely parseable (Distribution Tests).

Freshness

Stateless stages add the least latency of anything in a streaming pipeline: no buffering, no waiting for a watermark, no window to close. Latency is serialisation plus the function itself.
The exception is any per-record external call, which converts this stage's latency into somebody else's p99 and makes it the pipeline's slowest link (Tail Latency: Why p50 Being Fine Does Not Help).
Because a stateless stage never waits for completeness, its output is exactly as complete as its input — it neither improves nor degrades the freshness contract, which is a genuinely useful property when reasoning about the whole chain.

When the schema or meaning changes

This is the stage that absorbs upstream schema change, which is precisely why it should tolerate unknown fields rather than reject them. Reject on the fields you *use*, ignore the ones you do not (Backward Compatibility).
Adding a field to the output is a compatible change for downstream consumers reading by name. Renaming or retyping one is not, and streaming makes it worse than batch because there is no daily boundary at which someone would notice (Breaking Schema Changes).
A change to a filter predicate is a semantic change to every downstream metric, and it will not show up in any schema check. Version the predicate and record which version produced which range of output (Semantic Changes).

How to re-run this safely

Rewind the consumer and reprocess. Because the transformation is deterministic, the reprocessed output is byte-identical, so a downstream idempotent sink absorbs the replay with no correction step (Idempotent Data Pipelines).
Records rejected to a dead-letter destination are recoverable by fixing the parser and replaying that destination — but only if you kept the raw bytes rather than the exception message.
Records dropped by a filter are not recoverable from the output. They are recoverable from the log, for as long as retention holds, which is the argument for landing raw before filtering (The Raw Landing Zone).

What can go wrong

Failure modes

A poison record that raises on every attempt, stalling one partition while the others proceed — so throughput drops by roughly a partition's share and no alert fires (Head-of-Line Blocking).
A filter that stops matching after an upstream change and silently lets everything through, or matches too much and silently drops a category.
A flatMap that fans out more than expected, multiplying volume downstream and changing the grain without any declaration that it did (Grain: What Does One Row Represent?).
An enrichment lookup that returns null on failure instead of failing, so the output contains a growing share of records with a missing dimension and every group-by gets a swelling "unknown" bucket.
The mitigation failing: a dead-letter route with no alert and no owner is a queue that fills up with the data you needed and is never read.

Misreads

"Stateless means no side effects." It means no *retained* state between records. A stateless operator that writes to an external system on every record has side effects, is not replay-safe, and needs the same idempotency thinking as any sink (Idempotency Keys: The Mechanism).
"Enrichment from a cached lookup table is stateless." The cache is state: it has a size, a staleness, and a warm-up period after every restart, and the job's output depends on when it ran. Call it what it is (Stream Joins).
"Deduplication is a filter, so it is stateless." Deduplication requires remembering what you have already seen. It is the canonical stateful operator wearing a filter's clothes (Deduplication).
"If it is stateless, I do not need to think about ordering." True per record, false for the stage as a whole: a flatMap that emits several records per input, or a parallel stage that writes to a shared sink, can interleave in ways that matter to whoever reads the output (Webhook Ordering: Assume None).

Operating it

How you see it in production

Records in, records out and records rejected per operator, as three counters with one alert on the identity between them.
A histogram of per-record processing time for any stage doing real work, because a stateless stage that is slow is almost always doing I/O it should not (Histograms: A Distribution You Can Afford to Keep Forever).
The distribution of the enrichment result — specifically the share of records falling into the null or "unknown" bucket, which is the earliest visible sign of a broken lookup (Volume Anomalies).

What changes at 10x and 100x

At 10x, add parallelism up to the partition count. There is nothing else to think about, which is the entire selling point of this category.
At 100x, the partition count itself becomes the constraint and repartitioning the topic becomes necessary — which is not a stateless operation upstream, because changing partition count breaks per-key ordering for keys that move (Topics and Partitions).
Key cardinality is irrelevant here, and saying so is worth doing explicitly: it is the only stage in this module for which that is true.

What drives cost here

Cheap and predictable: cost scales linearly with records and with payload size, with no cardinality term and no retention term. This is the only part of a streaming pipeline where "more data costs proportionally more" is actually true.
The lever that matters is how early you drop bytes. Filtering and projecting before a shuffle removes serialisation, network and deserialisation cost from every stage after it (The Shuffle).
The one way to make a stateless stage expensive is a per-record remote call, which replaces a cheap CPU-bound function with a network round trip and forces parallelism up to compensate (What Serialization Costs).

What this approach costs

Splitting stateless work into its own job buys independent deployment and scaling and costs an extra hop: another topic, another serialisation round trip, another thing to monitor, and a small addition to end-to-end latency.
Tolerating unknown fields buys resilience to upstream change and costs you the early warning that a rejection would have given. The resolution is to tolerate but *count* — accept the record, increment a counter for the unexpected field (Contract Enforcement).
Filtering early buys cost and latency everywhere downstream and costs you the ability to answer questions about what was filtered, unless the raw stream is retained separately.

Where this applies

Almost nothing here is universal. These labels say what each claim is specific to, and where a different engine, format, warehouse or scale would differ.

GENERALFilter, map and flatMap over a record stream are the same three operators in every engine and every language binding; the properties that follow — determinism, restartability, rescalability — follow from the definition rather than from any implementation.
ENGINE-SPECIFICEngines differ in whether adjacent stateless operators are fused into one task (avoiding serialisation between them) or run as separate shuffled stages. Flink chains operators by default and Spark pipelines within a stage, so an apparently free extra map is free in one context and a network hop in another.

Where the depth lives

This domain teaches how data moves and how you know it arrived intact. It hands the rest off by name.

Backendevent-consumers dead-letter-queues serialization-cost idempotency-keys stateless-services

API Designwebhook-ordering

Observabilitytail-latency histograms

Concurrencydeterminism immutability pipeline-parallelism

OS & Networkinghead-of-line-blocking

Domains that do not exist yet

— DevOps / Production Engineering owns the deployment side of this: a stateless stage is the only part of a streaming platform that can use ordinary rolling-deploy machinery, because there is nothing to drain and nothing to migrate.