Distributed Data Processing on Ephemeral MicroVM Fleets

There is a version of this post that goes: microVMs boot in 179 milliseconds, therefore your Spark cluster should be ephemeral. I am not writing that one, because the interesting part of running distributed data work on a fleet of disposable VMs is not the boot time. It is which half of the workload survives contact with the network model, and that half is smaller than a vendor would like to admit. So: how to do it, what it buys you, and the structural reason a wide shuffle will not work. Better to learn that here than from a job that has been at 97% for forty minutes.

A fixed cluster is the wrong shape for bursty analytics

The standing-cluster model is a hangover from when provisioning a machine took a purchase order. You size for the peak, run continuously, and accept an average utilisation somewhere between embarrassing and undiscussed. Autoscaling improves the picture without fixing it: scale-up arrives after the queue has formed, and scale-down is deliberately timid so you do not thrash.

For bursty analytics the arithmetic is unkind. If the real demand is forty minutes of heavy fan-out at 06:00 and near-nothing until the next morning, a cluster that exists for twenty-four hours spends most of its life as a very expensive heartbeat. The workload wants a shape it cannot have: hundreds of workers for forty minutes, and zero for the other twenty-three hours.

Ephemeral microVMs give you that shape, and something the cluster never did: every run starts from a byte-identical environment. Anyone who has debugged a job that only fails on worker seven, because worker seven was rebooted in March and picked up a different numpy, knows what that is worth. The billing follows the shape too — $0.054 per vCPU-hour and $0.0162 per GiB-hour, CPU charged by seconds actually burned rather than by allocated wall-clock, and nothing at all between runs. That is the easy part of the argument. Here is the hard part.

The real win is resolving dependencies once, not N times

Ask what the expensive part of standing up a worker is and people say boot. It is not — boot on a snapshot-restore platform is a couple of hundred milliseconds. The expensive part is the dependency closure: pip resolving a scientific stack, a wheel built from source because no manylinux artefact matches your combination, a JVM pulling its dependency tree from an artefact repository, then the first import of a fat native library paging tens of megabytes off disk.

That work is identical for every worker, takes tens of seconds to minutes, and the standard ephemeral-worker design pays for it once per worker. Sixty-four workers doing ninety seconds of pip each is an hour and a half of machine time spent arriving at the same filesystem.

Forking is what deletes that. Do the resolution once in a parent guest, import the libraries so they are resident in RAM and not merely present on disk, and then fork. A fork inherits the parent's memory and filesystem copy-on-write, so each child wakes up in a process image where the install has already happened and the import cache is already warm. Mechanically, fork_tree snapshots the parent exactly once and boots the children from that snapshot in parallel, at roughly 200 to 500 milliseconds each. The per-call child count is capped at sixteen, so a wider fan-out is a loop over batches.

How workers find the driver, and the rule that decides everything

Now the network model, because every downstream decision falls out of it. Each sandbox gets its own Linux network namespace, veth pair and /30 carved from a 10.200.0.0/16 pool — 16,384 slots per host, pre-built and parked on a free list so allocation costs about five milliseconds rather than the hundred a cold build takes. Inside that design sits one rule that decides whether a distributed framework can run at all.

# The rule that decides your architecture. From the per-sandbox netns setup:
#
#   ensureRootFirst("FORWARD", "-s", PoolCIDR, "-d", PoolCIDR, "-j", "DROP")
#
# rendered on the host as, roughly:
iptables -I FORWARD 1 -s 10.200.0.0/16 -d 10.200.0.0/16 -j DROP

# 10.200.0.0/16 is the whole NATID veth pool -- every sandbox on the host lives
# on a /30 carved out of it. The DROP is inserted FIRST so it wins over the
# egress ACCEPT rules that follow. Consequence: sandbox A cannot open a socket
# to sandbox B, on this host or any other. That is a security control (without
# it a tenant could scan the /16 and hit a neighbour's SSH or Postgres), and it
# is not tunable per workload.
#
# The second surprise, from the same design: the guest IP is baked into the
# template snapshot and is IDENTICAL in every sandbox restored from it. The
# netns exists precisely so that thousands of guests can all believe they are
# 172.20.x.y without colliding. Any cluster protocol whose membership model is
# "each node advertises its own address to the others" is therefore starting
# from a false premise.

Read that twice before planning a cluster. Sandbox-to-sandbox traffic is dropped at the host's FORWARD chain, first rule, above the egress ACCEPTs, because without it one tenant could sweep the pool /16 and reach a neighbour's SSH or Postgres. It is a tenancy boundary, not a tunable. So there is no east-west network: workers cannot form a mesh, and two guests forked from the same parent onto the same physical host milliseconds apart cannot exchange a packet directly.

What they can do is egress through the host's NAT, and be reached from outside on a per-sandbox preview URL of the form `https://<port>-<sandbox-id>.<suffix>`, which routes through the control-plane edge into the guest and does preserve WebSocket upgrades. That is an HTTP path, not raw TCP, and every packet takes a proxy hop. A fine control channel; not a data fabric.

Which leads to the architecture that actually works: the driver is not a sandbox. The driver is your own process — a laptop, a CI job, a small always-on service — holding an API client, and the workers are things it drives over that API. Coordination goes through the control plane; data goes through object storage. There is no cluster membership protocol, because there is no cluster.

Shuffle is the thing that hurts

Here is the honest section. A wide shuffle — a big join, a group-by on a high-cardinality key, a repartition — is defined by every worker needing to read a slice of every other worker's output. It is an all-to-all exchange, and it has two hard requirements: fast worker-to-worker networking, and a lot of local scratch to spill into.

An ephemeral fleet with per-sandbox NAT provides neither. The first is not merely slow, it is prohibited by the FORWARD rule above. You can route a shuffle through object storage — broadly what the disaggregated-shuffle services attached to serverless Spark do — but those are purpose-built systems with careful attention to request amplification, and hand-rolling the idea onto a fleet of NAT'd VMs gets you a job dominated by round trips to a bucket. The second requirement, local scratch, is bounded by the disk the template baked and is thrown away when the guest dies.

A subtler problem catches people before they even reach the shuffle. Because the guest's network identity is frozen at bake time, every sandbox restored from a template presents the same guest IP and MAC — the namespace exists so thousands of guests can all be 172.20.x.y without colliding. Any framework whose membership model is "each executor registers, advertising the address peers should fetch from" starts from a false premise: the addresses are identical, and none is reachable anyway. Spark's block manager, Dask's default TCP peer addressing and Ray's object-store transfers all assume otherwise.

If your job's critical path is an all-to-all exchange, an ephemeral microVM fleet is the wrong tool and no amount of configuration will change that. Use a warehouse for the join and the fleet for the map.

The one framework worth a second look is Dask, which ships a WebSocket transport specifically so scheduler and worker traffic can traverse an HTTP proxy — the shape of path a preview URL provides. Check its current documentation before building on it, and note that even there the bag, delayed and per-partition workloads are what survive; the dataframe shuffle still will not be good.

Executor sizing: guest RAM is baked into the snapshot

Every distributed framework wants you to tune executor memory, and here that knob is not where you expect. Firecracker cannot change vCPU count or RAM at snapshot restore, so size is a property of the frozen machine: the create path reads the template metadata and forces the request to match before anything persists, because a per-request override would otherwise flow straight into the API response, the database row and, worst, the billing event. Practically, memory is a template property. The first-party base template bakes at 4 GiB and 8 vCPU, so a base-template worker is a 4 GiB worker whatever your job config says. Sixteen-GiB executors mean baking a 16 GiB template.

The second half of this constrains how wide you can go, and it is where a fan-out actually fails. Each host runs an admission gate over guest memory, with a budget of MemTotal minus a reserve (1 GiB by default) times an overcommit factor. Committed accounting charges the full baked size of every live guest, which caps a 32 GiB host at about seven 4 GiB sandboxes while the host sits nearly empty — Firecracker faults pages lazily, so a merely-alive guest holds a couple of hundred megabytes of real RAM. Working-set accounting measures actual residency instead and charges a create 25% of its baked size with a 512 MiB floor, so a 4 GiB base guest books 1 GiB rather than 4. Managed databases are always charged committed; everything else is squeezable.

A worked example: resolve once, fan out per partition

The driver holds the client, forks a batch of workers from one warm parent, hands each exactly one partition, and reduces the small summaries that come back. No worker talks to another; none needs to.

"""Partition-per-worker fan-out over an ephemeral microVM fleet.

The driver is THIS process -- your laptop, a CI job, a small always-on box.
It is deliberately NOT a sandbox. Workers never dial each other and never
dial the driver: guest-to-guest traffic is dropped at the host firewall by
design, so the only coordination channel is the API client you already hold.
"""
import json
from concurrent.futures import ThreadPoolExecutor
from pandastack import Sandbox

PARTITIONS = [f"s3://my-bucket/events/part-{i:05d}.parquet" for i in range(64)]
FANOUT = 16                       # fork_tree is capped at 16 children per call

# 1. ONE warm parent resolves the dependency closure. This is the whole point
#    of the exercise: pip resolution, wheel builds and the first import of a
#    fat native library cost tens of seconds and are byte-identical for every
#    worker. Doing it 64 times is the waste we are here to delete.
parent = Sandbox.create(
    template="base",
    ttl_seconds=3600,
    metadata={"kind": "partition-fanout", "role": "parent"},
)
parent.exec("mkdir -p /work/out", check=True)
parent.exec(
    "pip install --no-cache-dir 'pyarrow==17.0.0' 'pandas==2.2.2' boto3",
    timeout_seconds=900,
    check=True,
)
parent.filesystem.write("/work/task.py", open("task.py").read())

# Import once BEFORE the snapshot. A fork inherits the parent's RAM, so the
# children wake with the module already resident -- not merely present on disk.
parent.exec("python3 -c 'import pyarrow, pandas'", timeout_seconds=180, check=True)


def run_partition(pair):
    child, uri = pair
    try:
        r = child.exec(f"python3 /work/task.py {uri}", timeout_seconds=1800)
        if r.exit_code != 0:
            return {"uri": uri, "ok": False, "stderr": r.stderr[-2000:]}
        # The worker writes one small JSON summary. If what comes back here is
        # large, you have a shuffle, and a shuffle is the wrong shape for this.
        return json.loads(child.filesystem.read("/work/out/summary.json"))
    except Exception as err:
        return {"uri": uri, "ok": False, "error": str(err)}
    finally:
        child.kill()              # free the host RAM before the next batch


results = []
for start in range(0, len(PARTITIONS), FANOUT):
    batch = PARTITIONS[start:start + FANOUT]

    # 2. fork_tree snapshots the parent ONCE and boots the children from that
    #    snapshot in parallel. Children land on the SAME agent as the parent,
    #    which is why batch width is bounded by one host's memory budget rather
    #    than by the fleet's.
    children = parent.fork_tree(
        count=len(batch),
        metadata={"kind": "partition-fanout", "batch": str(start // FANOUT)},
    )

    with ThreadPoolExecutor(max_workers=len(batch)) as pool:
        results.extend(pool.map(run_partition, zip(children, batch)))

parent.kill()

# 3. The reduce happens HERE, in one process, over 64 small summaries. If your
#    reduce does not fit in the driver, you do not have this workload -- you
#    have a shuffle, and you want a warehouse.
failed = [r for r in results if not r.get("ok", True)]
total = sum(r.get("rows", 0) for r in results)
print(f"{len(results) - len(failed)}/{len(results)} partitions, {total} rows")

And the worker side, where the shared-nothing discipline becomes a property of the code rather than an aspiration.

#!/usr/bin/env python3
# task.py -- runs INSIDE one worker guest. One partition in, one summary out.
# Deliberately shared-nothing: it reads one object, writes one object, and
# never assumes another worker exists.
import json
import sys

import pyarrow.dataset as ds
import pyarrow.compute as pc

uri = sys.argv[1]

# Credentials are per-worker and scoped to this partition's prefix. There is no
# shared mount and no cluster-wide role -- a worker that is compromised has the
# blast radius of one object, not of the whole lake.
table = ds.dataset(uri, format="parquet").to_table()

# A per-partition UDF: the kind of work that genuinely parallelises with no
# cross-worker communication at all.
filtered = table.filter(pc.greater(table["amount"], 0))
summary = {
    "ok": True,
    "uri": uri,
    "rows": filtered.num_rows,
    "sum_amount": pc.sum(filtered["amount"]).as_py(),
    "distinct_users": len(pc.unique(filtered["user_id"])),
}

with open("/work/out/summary.json", "w") as fh:
    json.dump(summary, fh)

Two things in there are load-bearing. `kill()` runs in a `finally`, because the resource you are contending for is host memory and a batch that leaks guests will fail the next batch rather than itself. And what crosses back to the driver is a small JSON summary, not data — the moment that payload gets large you have reinvented a shuffle through the least suitable channel available.

Which job shapes work, and which do not

Stated plainly, so nobody has to discover it mid-migration. These work:

• Embarrassingly parallel, map-heavy work. One input, one output, no communication: file-per-task processing, document conversion, per-file parsing and validation, feature extraction.
• Per-partition UDFs, where the reduce is a cheap concatenation or a sum of small summaries.
• Dask bag and delayed workloads: task graphs whose edges are small and whose leaves are independent. The sweet spot for this model, and for the fork-from-a-warm-parent trick.
• Untrusted or tenant-supplied transforms. The hardware boundary earns its keep here: a customer's Python runs in its own kernel with its own scoped credentials, and no shared worker holds the union of everybody's keys.
• Anything already implemented as a queue plus a worker pool where the workers never talk to each other. You have the shape; the fleet is a better substrate for it.

And these do not:

• Big joins and wide shuffles. All-to-all exchange needs east-west networking that is dropped at the host firewall, plus local scratch that a disposable guest does not have. This is not a tuning problem.
• Long-running interactive clusters. A cluster you attach a notebook to and keep warm all day is the standing-cluster model this exists to avoid; you would be paying for machinery whose main benefit, near-zero cost between runs, you just amortised into irrelevance.
• Anything needing GPUs. Sandboxes are CPU-only with no device passthrough. If your pipeline is GPU-bound end to end, this is the wrong infrastructure, and I would rather say so now than in a support thread.
• Frameworks whose membership protocol requires peer addressing. Ray in particular spreads object-store and gRPC traffic across many inter-node ports; no configuration makes that work here.
• Jobs whose working set exceeds one baked template's RAM. Memory is a template property, so a bigger executor means a new template, not a flag.
• Iterative algorithms with tight synchronisation: an all-reduce per iteration is an east-west workload wearing a different hat.

What I would actually build

If the whole job is map-shaped, do not run a framework at all. A framework's value is scheduling, fault tolerance and shuffle. You have a scheduler already — the loop in your driver — and fault tolerance for an idempotent per-partition task is a retry. Shuffle is the part that does not work. Paying a framework's operational cost to use the third of it a plain loop already gives you is a poor trade.

If the job is mostly map with a narrow reduce, do the map on the fleet and the reduce in the driver or in a database. Sixty-four partition summaries reduce fine in one process; sixty-four billion rows do not, and that is the signal to push the aggregation into a warehouse rather than to widen the fan-out. If the job genuinely needs a wide shuffle, use something built for it and run a hybrid — the fleet for the untrusted or bursty per-partition work, the warehouse for the join.

The order I would build it in is unromantic. Time one partition in one sandbox end to end: create, install, work, result read. Then build the parent-and-fork version and time it again; if the delta is small, your dependency closure was cheap and the fork machinery is not earning its complexity. Then push the width up a batch at a time until the memory gate returns a 507, and you will have learned your real fan-out ceiling — a per-host number, not a fleet-wide one.

None of this is a criticism of Spark, which is very good at the thing this model cannot do. It is an argument that most of what runs on a standing cluster is map-shaped, was never a shuffle, and is being charged twenty-four hours a day for the privilege of sitting next to one.