A Dag returns a 2 GB pandas dataframe from one task to the next through XCom, and loops over each day's files inside a single task, so one bad file reruns all of them. How would you redesign it?

XComs are for small values, such as a path, a count or a list of file names, not dataframes: by default they are stored in Airflow's metadata database, which suffers with large values. Write the data to object storage or a table and pass its location through XCom. For the files, use dynamic task mapping: one task lists the files, and expand() creates one task copy per file at run time, so each file succeeds, fails and retries on its own. Cap the fan-out with max_active_tis_per_dag, and keep in mind that expand() creates at most 1024 copies by default.

1. The redesign

from airflow.sdk import dag, task
import pendulum

@dag(schedule="@daily", start_date=pendulum.datetime(2026, 1, 1, tz="UTC"), catchup=False)
def partner_files():

    @task
    def list_files(data_interval_start=None) -> list[str]:
        day = data_interval_start.strftime("%Y-%m-%d")
        return list_keys(f"s3://inbound/{day}/")   # a list of paths: small

    @task(max_active_tis_per_dag=16, retries=3)
    def load_file(path: str, table: str) -> int:
        rows = copy_into(table, path)              # the data moves inside the warehouse
        return rows                                # a count: small

    @task
    def check_total(counts: list[int]) -> None:
        assert sum(counts) > 0

    counts = load_file.partial(table="raw.orders").expand(path=list_files())
    check_total(counts)

partner_files()

• partial() fixes what is the same for every copy (the table); expand() maps over what varies (the path).
• Each file is its own task instance, shown with its index in the UI: a bad file fails alone and is retried alone.
• max_active_tis_per_dag=16 keeps 300 files from taking every worker slot; it counts copies across all active runs of the Dag.

2. Why not XCom for the dataframe

If tasks must hand over larger values, an object storage XCom backend keeps them out of the database; it is still better for the data to stay in the warehouse and move there.

3. Limits to know

• [core] max_map_length, 1024 by default, caps copies per expand(). A day with 3000 files fails the mapping: batch the paths (100 per copy) or raise the limit knowingly.
• Several mapped arguments give a cross product: expand(path=[...300], table=[...3]) makes 900 copies.
• List the files inside a task, not at the top of the Dag file. Airflow parses Dag files often, and top-level calls to S3 would run on every parse.

What interviewers want: XCom for small values and object storage or tables for data, passing locations instead of data, expand and partial for one task per file, failures and retries per file, max_active_tis_per_dag, the 1024 default of max_map_length and batching past it, cross products, and no work at the top level of a Dag file.