Airflow Object Storage: The /tmp That Broke Task Determinism

Every distributed system learns the same lesson the hard way: local disk is a lie. Airflow runs each task on whatever worker has a free slot, and when a task writes to /tmp, it writes to that worker's local filesystem — not a shared one. The next task on another worker can't see the file, and the rerun on a third worker can't reproduce the output.

So your pipeline becomes nondeterministic. The same DAG, the same input, different results depending on which machine happened to run each task. Teams debug this for weeks, blaming the data, the timing, the pandas version — when the real culprit is a file path that exists on exactly one machine.

Airflow 3's object storage layer is the shared whiteboard: a uniform ObjectStoragePath API over S3, GCS, and Azure Blob, with operators built around staging files.

You still pass small values through XComs. Everything else — files, parquets, CSVs, model artifacts — belongs in object storage, addressed by URI.

/usr/local/airflow/tmp isn't scratch space. It's a bug generator. Move the bytes, keep the references.

1. Local Filesystem Traps on Distributed Workers

A worker's local disk is private property. With LocalExecutor everything runs on one machine and /tmp looks shared — which is exactly why the habit survives. The moment you move to Celery or Kubernetes, each task can land on a different worker, container, or pod, each with its own filesystem that dies with the machine.

Retries make it worse. A task that retries often runs on a different worker than the first attempt, so even files written and read by the same task aren't guaranteed to be visible to itself. The failure is intermittent, which makes it look like flaky data instead of broken storage.

Nothing in Airflow synchronizes worker filesystems. Shared NFS mounts are a maintenance nightmare and a single point of failure; the platform's answer is to not need shared state at all.

from airflow.decorators import task


@task
def extract_orders():
    # WRONG: worker-local path, invisible to every other worker
    with open("/tmp/orders.csv", "w") as f:
        f.write(download_orders())


@task
def load_orders():
    # WRONG: may not exist on THIS worker, may be stale elsewhere
    with open("/tmp/orders.csv") as f:
        df = pd.read_csv(f)
        write_to_warehouse(df)

2. Object Storage as the Shared Truth

Object storage is the shared whiteboard: a bucket everyone can read and write, addressed by URI, durable beyond any single worker's lifetime. S3, GCS, and Azure Blob all work the same way — you write an object at a key, and any process with credentials can read it back, from anywhere, at any time.

This is what makes tasks deterministic. Extract writes to s3://landing/orders/2026-07-31/orders.csv. Transform reads that exact URI. A rerun of the same interval reads the same object and produces the same output — no hidden state, no machine identity, no luck.

The durability is the second win. A worker that dies mid-task loses its local files; a bucket retains objects until a policy deletes them. The artifact outlives the run, which is what makes reruns, audits, and debugging actually possible.

3. Airflow 3 Object Storage Abstractions

Airflow 3 wraps the cloud filesystems behind one API: ObjectStoragePath. You get pathlib-style operations — read_bytes, write_bytes, mkdir, list — that work identically against S3, GCS, Azure Blob, and local paths, selected by URI scheme. The connection and its credentials come from the usual connection/secret-backend machinery.

The abstraction matters more than the syntax. A task written against ObjectStoragePath is portable: the same DAG can run against a local minio bucket in dev and production S3 in prod, with the connection handling the switch. No cloud-specific code in your tasks.

Keep object storage for objects. The metadata DB still holds XComs and run state; the bucket holds bytes. That split is the whole architecture.

Two details from the docs that trip people up. The connection id can ride in the URI itself — s3://aws_default@my-bucket/ — or be passed as conn_id=..., and the keyword argument always wins; connections resolve lazily, so creating an ObjectStoragePath at DAG scope is safe. Provider support is opt-in: s3 needs apache-airflow-providers-amazon[s3fs], gcs needs the google provider, and the file scheme works out of the box. In Airflow 3 the public import moved to airflow.sdk.

from airflow.decorators import task
from airflow.io.path import ObjectStoragePath


@task
def stage_feed(data_interval_end: str) -> str:
    src = ObjectStoragePath(f"s3://landing/feeds/{data_interval_end}.csv")
    dst = ObjectStoragePath(f"s3://staging/feeds/{data_interval_end}/feed.csv")
    dst.parent.mkdir(parents=True, exist_ok=True)
    dst.write_bytes(src.read_bytes())
    return dst.as_uri()  # push the URI, not the bytes

4. Staging Patterns: Upload, Transform, Download

The production pattern is three phases, each writing to a named prefix. Ingest tasks land raw source files under raw/, transform tasks read raw/ and write clean/ with deterministic keys, and load tasks read clean/ into the warehouse. Every phase's output is addressable, auditable, and replayable.

Deterministic keys are the discipline that makes it work. Key by data_interval_end — s3://warehouse/orders/2026-07-31/ — so a rerun writes and reads exactly the same objects as the original run. Never key by timestamp or run_id; those defeat rerun idempotency.

This also fixes downstream trust. When a warehouse team asks where a number came from, the answer is a URI chain: clean/2026-07-31/orders.csv produced by transform, fed from raw/2026-07-31/feed.json. Lineage you can point at.

Because ObjectStoragePath is built on fsspec, warehouse tooling can read staged parquet with Airflow's credentials — conn.register_filesystem(path.fs) in DuckDB reads the object directly, no boto3 or cloud SDK in task code.

from airflow import DAG
from airflow.decorators import task
from datetime import datetime

with DAG(
    dag_id="orders_pipeline",
    schedule="0 3 * * *",
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=1,
) as dag:

    @task
    def ingest(interval: str) -> str:
        src = ObjectStoragePath(f"s3://raw/orders/{interval}/feed.json")
        return src.as_uri()

    @task
    def transform(src_uri: str) -> str:
        raw = ObjectStoragePath(src_uri)
        clean = ObjectStoragePath(f"s3://clean/orders/{src_uri.split('/')[-2]}/orders.parquet")
        clean.parent.mkdir(parents=True, exist_ok=True)
        clean.write_bytes(normalize(raw.read_bytes()))
        return clean.as_uri()

    @task
    def load(clean_uri: str):
        run_load(clean_uri)

    load(transform(ingest("{{ data_interval_end }}")))

5. Pass URIs, Not Bytes

XComs live in the metadata DB, and the metadata DB is a relational database, not a file server. A 10MB JSON payload doesn't just bloat one row — it slows every query the scheduler, webserver, and UI run against that table. The earlier incident in this series froze the whole scheduler on exactly this mistake.

The rule is binary: small values, like URIs, counts, and flags, travel through XCom; everything else travels through object storage, and XCom carries only the reference. A transform task returns s3://clean/orders/2026-07-31/orders.parquet — 60 bytes — not the parquet.

This pays off twice. The DB stays fast, and the data is where debugging tools can reach it: you can diff two reruns' artifacts in the bucket instead of digging through task logs.

6. Cost and Lifecycle of Staged Objects

Object storage is cheap until it isn't. Standard-tier S3 runs roughly $0.023 per GB per month, so a 500GB staging prefix is about $11.50 a month — rounding error. But every nightly run adds objects, and without lifecycle rules the staging bucket grows forever: 500GB becomes 5TB, then 50TB, and now it's a real line on the bill.

Lifecycle rules are the answer, and they're boring to set up and critical to keep. Expire raw/ and staging/ prefixes after a few days, move cold partitions to cheaper tiers, and never delete clean/ data that warehouses still reference without a documented retention decision.

One trap: cleanup tasks in the DAG, not just lifecycle rules. A cleanup task that deletes objects referenced by an active run creates exactly the nondeterminism you removed. Prefer lifecycle rules on prefixes for automatic expiry, and keep a DAG-side cleanup only for explicitly temporary artifacts.

7. The Migration Playbook: From /tmp to Object Storage

The migration is mechanical, which is the point — determinism is a refactor, not an act of faith. Step one: inventory every open(), write, and file path in the codebase. Step two: replace file I/O with ObjectStoragePath operations against a staging bucket. Step three: key every artifact by data_interval_end. Step four: change XCom payloads to URIs. Step five: prove it.

Proof is the double-run: clear a data interval and rerun it twice. Row counts must match exactly, artifact keys must match exactly, and the second run must read the objects the first run wrote. If anything differs, the pipeline still has hidden local or time-keyed state.

Then delete the /tmp habits. Add a lint rule that flags open() in task code, and review every new DAG for URI-only artifact passing. The discipline is what survives, not the migration.