Airflow Object Storage: Fix the Shared /tmp Nightmare

Distributed workers share nothing. Not memory, not disk, and especially not /tmp. Every DAG that assumes otherwise works until the scheduler places two tasks on different machines.

One team's pipeline wrote intermediate files to a shared /tmp path. Some runs landed both tasks on one worker and passed; others split across workers and produced garbage. Same code, different answers.

Object storage ends the lottery. Write artifacts to URIs, pass pointers between tasks, and you'll get the same truth on every worker. Deterministic. Portable.

Local Filesystem Traps on Distributed Workers

Worker disks are private by architecture. The scheduler places tasks wherever capacity exists, evictions reschedule mid-run, and autoscaling replaces hosts beneath you. Any path starting with /tmp is a promise the platform never made.

You'll notice the trap in single-worker history. Dev runs everything locally, staging runs one worker, and early prod is small enough that tasks collide on hosts. The DAG looks correct for months, then the fleet grows and placement spreads.

Prove sharing is broken before it matters. Force consecutive tasks onto different workers on staging and watch /tmp handoffs fail on demand. A test that fails deterministically beats an incident that fails randomly.

The failure mode is always the same: /tmp looks shared on one box and isn't on two. One worker writes, the scheduler schedules the reader elsewhere, and the file simply isn't there. Object storage ends the guessing — one URI both tasks can see, no affinity rules, no NFS prayers.

Object Storage as the Shared Truth

Object storage is the shared truth every worker can reach. Buckets persist beyond pods, replicate across zones, and serve identical bytes to any host with credentials. Tasks stop caring where they run.

You'll access it through connections, not SDK config. The conn_id selects credentials per environment, IAM scopes them to the artifacts bucket, and rotation happens in one place. Prod and dev differ by connection, never by code.

Install the fsspec extra for your backend. S3 needs the amazon provider's s3fs extra, GCS its gcsfs equivalent. Without it the URI scheme has no driver and every path operation fails the same way.

pip install 'apache-airflow-providers-amazon[s3fs]'  # fsspec backend for S3

# scoped writer connection (least privilege on the artifacts bucket)
airflow connections add aws_reports \
  --conn-type aws \
  --conn-login "$AWS_ACCESS_KEY_ID" \
  --conn-password "$AWS_SECRET_ACCESS_KEY" \
  --conn-extra '{"region_name": "us-east-1"}'

# smoke test from any worker
python3 -c "from airflow.sdk import ObjectStoragePath; b = ObjectStoragePath('s3://aws_reports@acme-artifacts/'); b.mkdir(exist_ok=True); print('bucket OK:', b)"

Airflow 3 Object Storage Abstractions

ObjectStoragePath feels like pathlib because it is modeled on it. Slash joins segments, open streams bytes, iterdir lists, mkdir ensures prefixes. You'll learn four methods and port the pattern across providers.

The URI carries everything. Scheme selects the backend, conn_id selects credentials, path selects the object. Swap s3 for gs plus a GCP conn_id and identical task code runs on the other cloud.

Serialize paths, not handles. Tasks return the URI string; downstream tasks rebuild the path object with their conn_id. Strings cross XCom cleanly while live handles cannot.

Import from the right place on Airflow 3: from airflow.sdk import ObjectStoragePath (the old airflow.io.path location moved). S3 needs pip install apache-airflow-providers-amazon[s3fs] — it pulls aiobotocore, which is deliberately excluded from the default install to dodge botocore conflicts, so missing-s3fs is the first thing to check when S3 paths won't resolve. The URI shape is protocol://[conn_id@]bucket/key: the protocol picks the backend (s3, gs, azure), the userinfo slot carries the conn_id, or pass conn_id= explicitly (explicit wins). Omit it and Airflow falls back to that backend's default connection.

Create paths freely at DAG top level — connections resolve lazily on first use, not at import, so defining base = ObjectStoragePath(...) globally never opens a network call during parsing. Paths serialize cleanly (path, conn_id, kwargs round-trip), which is exactly how you pass them through XCom between tasks without pickling filesystems.

import pendulum
from airflow.sdk import ObjectStoragePath, dag, task

base = ObjectStoragePath("s3://aws_reports@acme-artifacts/sales/")

@dag(
    dag_id="sales_stage",
    schedule="0 6 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["sales", "object-storage"],
)
def sales_stage():
    @task
    def extract(logical_date: str | None = None) -> str:
        import pandas as pd

        base.mkdir(exist_ok=True)
        path = base / f"raw_{logical_date}.parquet"
        df = pd.DataFrame({"order_id": [1, 2, 3], "amount": [19.0, 42.0, 7.5]})
        with path.open("wb") as f:
            df.to_parquet(f)
        return str(path)  # pointer travels, bytes stay in the bucket

    @task
    def transform(path: str) -> str:
        import pandas as pd

        src = ObjectStoragePath(path, conn_id="aws_reports")
        with src.open("rb") as f:
            df = pd.read_parquet(f)
        out = base / ("clean_" + src.name)
        with out.open("wb") as f:
            df.dropna().to_parquet(f)
        return str(out)

    transform(extract())

sales_stage()

Staging Patterns: Upload, Transform, Download

The staging pattern has three steps. Upload raw artifacts under a dated prefix, transform by reading URIs and writing new URIs, and let consumers download only what they need. Each step is independently retryable because inputs are immutable objects.

You'll namespace by DAG, date, and stage. sales/2026-09-04/raw_ plus clean_ prefixes keep runs isolated and reruns idempotent: rewriting the same keys converges instead of duplicating. Debugging starts with listing one prefix.

Chain through return values. TaskFlow passes the URI string downstream implicitly, so the graph reads extract to transform to load with no manual XCom calls. The lineage is the path itself.

Lean on the pathlib shape: / joins keys, mkdir(exist_ok=True) ensures prefixes, open('wb'/'rb') streams bytes, stat() gives st_size/st_mtime plus backend extras (ETag, ContentType on S3 — don't rely on those cross-backend). copy/move follow fsspec semantics, including remote-to-remote streaming that walks the tree file by file; same-store copies stay optimized. Need DuckDB, Iceberg, or pandas on the other end? Grab path.fs — the fsspec filesystem authenticated from your conn_id — and hand it over: conn.register_filesystem(path.fs) lets DuckDB read_parquet straight off the path with Airflow's credentials. Custom store? Register once with attach() at DAG top level so every task reuses it.

Pass URIs, Not Bytes

Pass URIs, not bytes, between every task. The pointer weighs a hundred bytes; the parquet weighs a hundred megabytes. XComs stay tiny, the metadata DB stays fast, and workers fetch directly from the bucket at full speed.

You'll enforce a size rule in review. Anything over a few KB travels by reference, no exceptions. The one team that allowed small frames through XCom watched small become 10MB within a quarter.

Downstream engines join the pattern. DuckDB, pandas, and Iceberg all accept fsspec filesystems or URIs, so analysis tasks reuse Airflow credentials instead of minting their own. One credential path, every engine.

import duckdb
from airflow.sdk import ObjectStoragePath, dag, task
import pendulum

@dag(
    dag_id="sales_analyze",
    schedule="0 7 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["sales", "analytics"],
)
def sales_analyze():
    @task
    def summarize(path: str) -> str:
        src = ObjectStoragePath(path, conn_id="aws_reports")
        conn = duckdb.connect(database=":memory:")
        conn.register_filesystem(src.fs)  # reuse Airflow creds in duckdb
        total = conn.execute(
            f"SELECT sum(amount) FROM read_parquet('{src}')"
        ).fetchone()[0]
        return f"total={total}"

    summarize(path="s3://aws_reports@acme-artifacts/sales/clean_raw_2026-09-04.parquet")

sales_analyze()

Cost and Lifecycle of Staged Objects

Storage bills what you forget. Staging prefixes accumulate run after run, each harmless alone, each billed forever together. One team's bucket tripled in a quarter with zero new pipelines.

You'll expire aggressively. Fourteen-day lifecycle rules on staging prefixes delete scratch automatically while promoted outputs version explicitly. Tag objects by producing DAG so chargeback shows which team funds the growth.

Audit quarterly with one listing. Sort prefixes by size, name the top three owners, and ask each whether the data earns its rent. Deletion is the only optimization that compounds.