Dynamic Task Mapping: Graphs Expanded at Run Time

A workflow defined with a fixed number of tasks can be validated, visualised, and reasoned about before it runs. That property is also its limit: the number of input files, table partitions, or shards is usually not known until the run starts, and a graph that hard-codes twenty tasks either breaks when there are twenty-one or wastes nineteen slots when there are two. Dynamic task mapping defers the width of the graph to run time, at the cost of losing most of the static guarantees the scheduler was built on.

Expansion at the moment of readiness

The mechanism is a task that, instead of doing work, produces a list. Its successors are not defined in the graph; they are created when the list is known, one per element, each carrying an index and the element’s value. In a scheduler with a persistent run state, this means writing new task instances into that state after the run has started, and the downstream graph must be able to reference tasks that did not exist at parse time.

The parse-time graph therefore contains a single stand-in node for the expanded region. The scheduler validates the static part, and the dynamic part is validated only in its shape: a mapped task must be followed by something that can accept a variable number of predecessors, and the scheduler must know which downstream tasks are affected by the expansion.

What the list may contain is constrained. If the list is derived from the output of a previous task, the expansion happens at run time and depends on that task’s result. If it is derived from a fixed configuration, it could have been static. The useful case is the first, and it introduces a dependency the scheduler cannot check: whether the produced list is empty. An empty list expands to zero tasks, and a downstream task that requires at least one input is now unsatisfiable. The scheduler must decide whether zero-width expansion is a valid outcome, as it is for a filter-like map, or an error, and that decision is a property of the downstream task, not of the mapping.

Keying the expanded instances

Each expanded instance needs an identity that is stable across retries and re-runs, because the scheduler stores per-instance state and must be able to match a completion to the instance that produced it. The index in the list is the obvious key and it is not stable: if the upstream task produces the list in a different order on a second run, whether because it read a directory whose listing order changed or because it parallelised its own query, index 3 refers to a different element.

The fix is to key on the element’s content rather than its position: a hash of the value, a name derived from the value, or an explicit key that the upstream task supplies alongside each element. Then the identity of an instance is a function of what it processes, and a re-run that produces the same elements produces the same keys in a different order, which the scheduler can reconcile.

The cost is that two elements with the same content collapse into one instance. If the list legitimately contains duplicates, such as two files with identical names in different directories, the key must include the distinguishing part, and the upstream task is responsible for constructing a key that is unique within the expansion. A scheduler that keys only on position pushes this problem onto the operator, who sees an instance fail to match and has no way to correct it.

import hashlib, json

def instance_key(element, occurrence):
    """Stable per-element key; occurrence disambiguates true duplicates."""
    body = json.dumps(element, sort_keys=True).encode()
    return f"{hashlib.sha256(body).hexdigest()[:16]}-{occurrence}"

The occurrence counter is needed because content hashing alone cannot distinguish identical elements. It must be assigned deterministically, by sorted position among equals rather than by arrival order, or the key is unstable again for exactly the case it was meant to fix.

The reduce side

Fan-out is the easy half. The hard half is the task that consumes all the expanded instances, because its readiness condition is now over a set whose size was unknown when the graph was written.

Three shapes are possible. The consumer may require all instances to succeed, which is the default trigger rule applied to a variable predecessor set: the scheduler counts the expanded instances and waits for that many successes. The consumer may require any, which means it can start as soon as one instance succeeds. The consumer may be absent, with each instance writing independently to its own partition, in which case the run is complete when all instances are terminal and there is no join at all.

The first shape has a failure mode specific to dynamic graphs: the count is fixed at expansion time, but the set of instances can change if an instance is retried and the retry produces a different result, or if the scheduler adds instances on a re-run. A consumer that cached the count will wait for a number that no longer matches. Recomputing the count from the run state at each evaluation avoids this, at the cost of a query per evaluation.

The second shape, starting on any success, combined with a mapped region is the case where partial results are most dangerous. The consumer runs with some fraction of the input, produces an output that looks complete, and the remaining instances finish later with nowhere to put their results. Making the consumer aware that it is reading a partial set requires it to know the total, which is precisely the information the dynamic graph has and the consumer’s data does not.

Concurrency limits and the expanded width

An expansion of ten thousand instances is not a graph with ten thousand runnable tasks; it is a graph with ten thousand tasks of which a few may run at once. The scheduler’s concurrency limit applies to the expanded set as a whole, and the practical problem is the size of the run state: ten thousand instance records, each with its own attempts, is a table that must be written before any work happens and queried on every readiness evaluation.

Schedulers handle this in two ways. One is to expand lazily, creating instances in batches as slots free up, so that the state grows with the work rather than with the plan. The other is to cap the expansion and require the upstream task to produce a list of at most some size, pushing the batching into the task’s own logic, at one instance per hundred files for example. The first keeps the graph faithful to the data and puts the burden on the scheduler; the second keeps the scheduler simple and puts the burden on the task author to choose a batch size that balances per-task overhead against the width of the graph.

The choice matters because per-task overhead in a scheduler is not zero. A task dispatch involves a durable write, a queue message, a worker process start, and a state transition, and for a task that runs for a few seconds the overhead can exceed the work. Batching a hundred small units into one task amortises that overhead a hundredfold, at the cost of coarser retry granularity: a batch that fails retries the whole batch, including the ninety-nine units that succeeded.

What static analysis loses

A statically defined graph can be checked for cycles, for unreachable tasks, for trigger rules that can never be satisfied, and for a critical path whose length is known before execution. A dynamically expanded graph can be checked for none of these beyond the static skeleton, because the expanded portion does not exist until run time.

The practical consequence is that errors move from definition time to run time, and a run-time error in an expanded region is more expensive: the run has already done work, the state contains instances that will never complete, and the operator sees a failure whose cause is a list produced by a task rather than a mistake in the definition. The mitigation is to validate the list inside the producing task, checking that it is non-empty if required, that keys are unique, and that its size is bounded, and to fail there, so that the failure is attributed to the task that produced the bad list rather than to the expansion mechanism.

The other loss is the ability to estimate the run’s cost. A static graph’s resource needs are the sum of its tasks’ known footprints; an expanded graph’s needs are a function of the data. Capacity planning for such a graph has to be based on the observed distribution of expansion widths rather than on the definition, and a single outlier, such as one day with a hundred times the usual number of files, can exhaust the pool that was sized for the median.

Where mapping belongs

The pattern is worth using when the units of work are genuinely independent, numerous, and discovered at run time: per-file parsing, per-partition compaction, per-shard validation. It is worth avoiding when the units are few and known, because the static form is easier to inspect, and when the units are not independent, because the expansion gives the illusion of parallelism over work that must be ordered.

The boundary between the two is the reduce step. A map with no reduce is a set of independent writes, and its correctness depends on each write being fenced and idempotent: the mechanisms in worker leases and zombie tasks and idempotent task boundaries. A map with a reduce is a distributed computation, and the reduce inherits every problem the scheduler’s own state machine has: partial failure, retries that change the input set, and a completion condition that is a function of a set rather than of a fixed list of predecessors. The mapping mechanism does not solve those; it only makes the set bigger.