DAG Workflows

Many studies on DesignSafe need several jobs, not one. A machine-learning study, for example, first runs a 75-simulation OpenSees sweep on 48 cores and then trains a regression on the sweep’s results. The training belongs in its own small job, because a retrain should cost one core rather than a repeat of the whole sweep. Separate jobs, however, put a person in the loop. The training job cannot start until the sweep finishes, so someone watches the queue and then submits the second job with the right paths. For one study that is a chore. For a campaign where several sweeps feed several analyses, it is unmanageable.

A workflow removes the person from the loop. You describe the jobs and how they depend on each other with dapi’s dapi.workflows, and DesignSafe’s workflow service runs the graph, submitting each job on your behalf the moment the jobs it depends on finish. Close the notebook mid-campaign and the run continues. Each job runs on the same queues and the same allocation as a job submitted by hand. The workflow replaces the human coordinator, nothing else.

A workflow is a graph of jobs¶

The sweep and train jobs, joined by one arrow, already form a workflow. The arrow points from sweep to train because train needs what sweep produces. For dapi it means two things. Train cannot start until sweep finishes, and train’s input is the archive sweep leaves behind.

Larger studies add more jobs and more arrows, and the arrows alone fix the schedule. Jobs joined by an arrow run in order. Jobs with no arrow between them run at the same time, so a study fans out across the machine with no extra effort from you. The one rule is that the arrows cannot form a loop, because a job cannot wait for its own results. This shape, jobs with one-way arrows and no loops, is called a directed acyclic graph, a DAG, and validate() checks the no-loop rule before anything is submitted.

Declare the graph¶

In a script, order comes from the order of the lines. In a workflow it does not. The graph contains only the dependencies you declare, and the order in which you add jobs means nothing. Declaring an edge takes one of two forms. Name the dependency directly with depends_on, or embed a reference to another task’s output, as the train job below does.

from dapi.workflows import Workflow, JobTask

wf = Workflow("opensees-ml")
sweep = wf.add(JobTask("sweep", job_dict=sweep_job))

train_job["fileInputs"][0]["sourceUrl"] = sweep.output(
    "archive_uri", suffix="/inputDirectory"
)
train = wf.add(JobTask("train", job_dict=train_job))

wf.validate()                 # rejects cycles, duplicate ids, unknown refs
results = wf.run(ds)

sweep.output(...) is a reference, not a value. When you embed it in the train job, you point train’s input at sweep’s future archive and declare the edge sweep -> train in one move. dapi calls this an output reference. dapi can replace the reference with a real path before it submits anything, because dapi gives every task of a run a fixed archive directory, <user>/dapi-workflows/<name>/<run_id>/<task> in MyData, so dapi knows where sweep’s results will land before sweep exists.

Watching a run¶

While run() waits, it streams a timestamped transition for every task. The transcript below records one run of the workflow above, submitted from a laptop.

[18:12:10] pipeline 'opensees-ml-20260810-181209': submitted (2 tasks)
[18:12:41]   task sweep: created -> active
[18:12:41]   task train: created -> pending
[18:19:19]   task sweep: active -> completed
[18:19:19]   task train: pending -> active
[18:33:05]   task train: active -> completed

The service submitted train the moment sweep completed, pointed at an archive that had existed for seconds. The stream is a window, not a leash; close the notebook and the pipeline keeps running, with results landing at the same archive paths.

How the scheduler decides¶

The scheduler maintains a ready set, the tasks whose dependencies have all finished. The scheduler submits independent tasks together and polls them in one pass, so parallel branches cost you nothing extra. When a task fails, the scheduler never submits its dependents, and tasks on other branches keep running. Step through a clean run, then a run where one shard fails.

Fan-out and fan-in¶

The stepper’s graph ran on Stampede3 as three Monte-Carlo pi shards (1 core each, different seeds) feeding one aggregator, which combined their counts and estimated pi to within 6.3e-4 from six million samples. Tapis apps such as python-s3 accept exactly one input directory, yet a fan-in needs every parent’s output. Because all tasks of a run archive under one run root, the aggregator’s single input directory is the run root itself, and because the scheduler submits the aggregator only after every parent has archived, Tapis stages the run root with every shard’s results already in it. This is the run-root pattern.

Archive volume, not compute, dominated the first run. The second run sets a Tapis archiveFilter on every task, so each archives only its result files.

Two lessons carry over to any workflow. Filter what each node archives, since archiving scales with data volume and the filter cut it sixteenfold. And budget a fixed staging delay for every directory input, since staging time did not shrink with the data at all.

When to fuse instead¶

A strictly linear graph gains no parallelism, yet each of its jobs still waits in the queue once. sequence_job() packs ordered steps into a single job on one node, with a generated fail-fast driver, and the fused job can itself be a node in a larger graph.

from dapi.workflows import sequence_job

fused = sequence_job(
    ds,
    steps=["python3 call_pylauncher.py", "bash ml_post.sh"],
    input_dir_uri=staged_uri,
    node_count=1, cores_per_node=48, max_minutes=45,
    queue="skx-dev", allocation=allocation,
)

Choose the graph when stages need different resources or fan out; choose the fused job when they share a machine and always run together.

Run these workflows yourself¶

Both shapes on this page ship as executed notebooks. The OpenSees ML workflow runs the sweep-feeds-training chain, and the pi fan-out workflow runs three parallel jobs into an aggregator. The fan-out exercise poses that build as four TODOs with hints, and the dapi workflows documentation covers the full API.

Custom containers as nodes¶

Tools that need their own software stack run as containers inside ordinary jobs, and a containerized job drops into a graph unchanged. Containers on HPC covers the delivery paths and the driver script. A gmprocess study then becomes a graph in which one node fetches waveforms, parallel container nodes process record shards, and a final node aggregates a report, each sized to its own resources.