How granular should a pipeline task be?

Every Airflow estate I have worked on has had a task called something like pre_merging. It is three hundred lines long. It calls two APIs, filters one of the responses by a date range, joins the two frames, reshapes the result into the layout the next stage wants, and writes it somewhere. It was written in one sitting by someone who understood the whole flow, and it worked on the first run.

Then one morning at 3am it fails with KeyError: 'security_id'. The log tells you which line raised. It does not tell you which of the two API responses lacked the column, whether the filter emptied the frame before the join, or whether the join itself produced the wrong shape. To find out, you rerun the whole forty minutes with print statements added, and watch it fail again at minute thirty-eight.

The metadata-driven ETL post argued about which behaviour belongs in a spec and which belongs in code. This post is about the code side: once a pipeline is Python, how big should each task be? There is a failure mode at each end, and I want to defend a rule that sits between them.

Two ways to get it wrong

The fused blob on the left is the pre_merging task. It has one virtue: it is fast to write. Everything else about it is a cost paid later. The intermediate data lives only in local variables, so nothing can be inspected after the run. A failure anywhere means a restart from the beginning. The retry policy is one number for five different kinds of work, so a flaky API call and a genuine logic bug get the same three attempts. And the task's name describes none of what it does, because no single name could.

The confetti pipeline on the right is the over-correction, and I have seen teams arrive there in direct reaction to the blob. Every groupby gets a task. Every column rename gets a task. The DAG graph has forty boxes for a job that does three things, and reading it is harder than reading the blob was, because the shape of the work has been dissolved into scheduler overhead. Each boundary also has a real price: the scheduler has to notice one task finished and queue the next, and the data has to be persisted and reloaded across every gap. Forty boundaries on a small frame can cost more wall-clock time than the work itself.

Both ends fail for the same underlying reason. The person cutting the boundaries was thinking about the code, not about the run.

The rule

Cut a task boundary where at least one of these is true:

1. You would stop and look at the data here. If, while debugging, you would want to open the output of this step and eyeball it, then the step should be a task and its output should be persisted where you can open it.
2. You would want a restart to begin here. If the work before this point is expensive, slow, rate-limited, or non-idempotent, and the work after it is where things usually go wrong, then a failure after the boundary should not redo what came before.

And nowhere else. If neither is true of a candidate boundary, the step is a function inside a task, not a task. Function-level granularity in Python is free; task-level granularity in Airflow is not, so it has to earn its place with one of the two reasons above.

Both halves of the rule matter. The first half is what kills the blob: pre_merging has at least four places where you would want to look at the data, and it exposes none of them. The second half is what kills the confetti: a column rename is not something you would ever inspect in isolation or restart from, so it does not get a box.

What it looks like

Take the pre_merging task and apply the rule. Here is where the boundaries land, and why each one is there:

• fetch_positions and fetch_constituents are separate tasks because they are separate restart points. Each is an external call with its own rate limit and its own failure profile. When the benchmark vendor is down, the positions call should not be repeated, and the retry count on each should reflect that system, not an average of both.
• filter_to_universe is a task because it is the first place I would look when the numbers are wrong. An empty frame after the filter is the single most common cause of a downstream KeyError, and I want to see the row count on its own line in the log, and the frame itself on disk.
• join_holdings_to_constituents is a task because the join is where the data model meets reality. Unmatched keys on either side are the thing I want to inspect, and the joined frame is the thing every later step depends on. It is also the natural restart point for everything downstream: if the reshape has a bug, I fix it and rerun from the join output, not from the API calls.
• reshape_for_attribution is a task because its output is the contract with the next pipeline, the one in the attribution post. A contract boundary is always an inspection boundary.

Notice what did not become a task. Inside filter_to_universe there is a date parse, a type coercion, and a rename. Inside the join there is a key normalisation. None of those are places I would ever stop, and none are worth a restart point, so they stay as small named functions inside the task. The task body reads as a short list of function calls, which is the level of granularity that belongs in code.

@task
def filter_to_universe(positions_path: str, run_date: date) -> str:
    positions = read_frame(positions_path)
    positions = coerce_position_types(positions)
    positions = positions[positions.as_of == run_date]
    positions = drop_zero_holdings(positions)
    log.info("positions in universe: %d rows", len(positions))
    return write_frame(positions, stage="filtered")

Three helper functions, one log line that carries the number I would check first, one persisted output. The task is small, but not because it does one thing; it is small because it does one inspectable thing.

Do the join yourself

One specific consequence of the rule is worth calling out, because it runs against a common shortcut. Many source APIs will do the join for you: ask the positions service for holdings with their benchmark constituent data attached, and you get one response and one fewer step.

I do not take that offer. The join is exactly the step where I most want to inspect both inputs and the result, and an API-side join gives me none of them. When a security is missing from the joined response, I cannot tell whether it was absent from holdings, absent from the benchmark, or dropped by whatever join semantics the API chose. Fetching both sides and joining in my own code costs one more task. It buys me both raw inputs on disk, the unmatched keys as a first-class artefact, and join logic I can read and change. Under the rule, the join earns its boundary on both counts, so it should be mine.

When every task is a container

A common objection: none of this applies if the work runs under DockerOperator or KubernetesPodOperator, because then the whole job is one container and Airflow only sees one box. I think that gets it backwards. Containers change the price of a boundary, not the rule.

The price goes up. A task boundary now costs an image pull and a container start, often tens of seconds on Kubernetes, and every intermediate has to go through object storage because the container's filesystem dies with it. So each boundary has to clear a higher bar, and "nowhere else" bites harder. The two questions still decide where the cuts go; you will simply answer yes to fewer of them.

The real risk is the one to name out loud: the fused blob just moves inside the image. Airflow shows one green task called run_job, and the three-hundred-line pre_merging lives inside it where no retry policy, no log line, and no graph view can tell its stages apart. That is the blob with better packaging, and it is the worst outcome of the three.

There are two honest ways to keep the rule with containers:

1. Same image, different command per task. Build one image, and have each Airflow task run one stage's entrypoint, with stage outputs landing in object storage. You keep per-stage retries, ownership, and observability, and pay the startup cost at every boundary. This is my default.
2. One container, checkpointed stages inside. The entrypoint runs the stages in sequence, persists each stage's output, and accepts a --from-stage flag so a rerun can resume from a checkpoint. Airflow sees one task. You get the inspect property and the restart property at near-zero boundary cost, and lose per-stage retry policy and the scheduler's view of where it failed. This is the escape hatch, for when startup cost is the dominant term or the stages share a large in-memory frame.

Stated generally: a boundary is any point where the output is persisted and the run can resume from it. An Airflow task is one implementation of that. A checkpointed stage inside a container is another. Choose the Airflow one when the stages need different retry, alerting, or owners, and the in-container one when the boundary cost outweighs those.

Where judgement still applies

The rule is a test, not a formula, and I want to be honest about where it leaves room for taste.

"Would I stop and look here?" depends on what has gone wrong in the past on this pipeline. A step that has never once been the culprit does not need its own box, even if it theoretically could be. Boundaries should follow the scars. A pipeline with two years of production history will have its boundaries in different places from the same pipeline on day one, and that is correct, not inconsistent.

"Would I restart from here?" depends on cost. On a frame of ten thousand rows, refetching from the API is cheap and the restart argument is weak. On ten million rows from a rate-limited vendor, it is the strongest argument in the design. The same code deserves different boundaries at different scales.

What I resist is turning either question into a number: no "every task under N lines", no "no more than M tasks per DAG". Those rules are easy to check in a linter and easy to satisfy without thinking. The two questions above cannot be satisfied without thinking, which is the point.

The smells, on each side

Signs you are on the blob side:

• A task name that is a verb phrase with "and" in it, or a name like pre_merging and process that avoids saying what it does.
• The only way to see intermediate data is to add a print and rerun.
• One retry count governs an API call and a pure transformation.
• Debugging always starts with "let me run the whole thing locally".

Signs you are on the confetti side:

• The DAG graph needs zoom to read, for a job with three real stages.
• Tasks whose output nobody has ever opened.
• Scheduler latency between tasks is a visible fraction of the run.
• Adding a column requires touching four task definitions.

Both lists share a diagnosis: the boundaries were placed by looking at the code rather than by asking where the run needs to be observed and resumed.

One thing this post leaves out

Every boundary produces an intermediate, and where that intermediate lives is a design decision of its own: XCom, a staging table, a file in object storage. That choice interacts with everything above, since a boundary you cannot inspect is not an inspection boundary at all. It is also a large enough topic to deserve its own post, so I have kept this one to the question of where to cut and left what crosses the cut for next time.

The takeaway

Task granularity is not a property of the code. It is a property of how you will debug and restart the run. Cut where you would want to look at the data, and where you would want a failure to resume from, and treat every other candidate boundary as a function. The blob and the confetti are both what you get when boundaries are drawn from the code's shape instead of the run's needs, and the fix for both is the same two questions asked at every proposed cut.