CWL interop with Dagster or other orchestrators

[davidelbert]

The workflow group here is more focused on workflow engines and the great work done on so many such options that help build workflows or run workflows that are well suited especially, perhaps, in HPC or generally computational tasks. I was interested in the rather simple idea of using CWL for sharing/interoperability of workflows to more general purpose workflow orchestrators. My group uses Dagster so decided to try to wrap CWL to run in Dagster. That would mean any CWL that is a DAG could run in Dagster and get the benefits of that. My first thought was to just wrap the whole CWL and I thought a design pattern would be:

# pyproject: dagster[catalog]
import json
import subprocess
from pathlib import Path
import dagster as dg

@dg.op(
    config_schema={
        "cwl_path": dg.Field(str, description="Path/URL to the CWL workflow"),
        "job_path": dg.Field(str, description="Path to the CWL input object (YAML/JSON)"),
        "outdir": dg.Field(str, default_value="outputs"),
        "cachedir": dg.Field(str, default_value=".cwl-cache"),
        "extra_args": dg.Field([str], default_value=[]),
    },
    required_resource_keys=set(),  # add resources for storage, s3 creds, etc., if needed
)
def run_cwl_op(context: dg.OpExecutionContext) -> dict:
    cfg = context.op_config
    outdir = Path(cfg["outdir"]).resolve()
    outdir.mkdir(parents=True, exist_ok=True)

    # Example with cwltool CLI (reference CWL runner)
    # You can swap this for "toil-cwl-runner" to use Toil’s backend.
    cmd = [
        "cwltool",
        "--outdir", str(outdir),
        "--cachedir", cfg["cachedir"],
        "--enable-dev",            # as needed
        "--no-match-user",         # as needed for container uid mapping
        cfg["cwl_path"],
        cfg["job_path"],
        *cfg["extra_args"],
    ]
    context.log.info(f"Running: {' '.join(cmd)}")
    proc = subprocess.run(cmd, check=True, capture_output=True, text=True)

    # cwltool prints a JSON object with resolved outputs to stdout
    outputs = json.loads(proc.stdout)

    # Emit each file-like output as a materialization and build a clean dict
    materialized = {}
    for name, val in outputs.items():
        # CWL ‘File’ objects have 'location' or 'path'
        if isinstance(val, dict) and val.get("class") == "File":
            p = val.get("path") or val.get("location")
            context.log.info(f"Output {name}: {p}")
            dg.AssetMaterialization(
                asset_key=dg.AssetKey([name]),
                metadata={"path": p},
            )
            materialized[name] = p
        else:
            materialized[name] = val

    return materialized

@dg.job
def cwl_job():
    run_cwl_op()

[davidelbert]

• I decided oncwltool as the runner since it if the reference CWL implementation and it is the only one I have ever used!
• Inputs and outputs.
  Put your CWL input object (YAML/JSON) under Dagster config control, or build it dynamically from upstream op outputs. Parse the runner’s JSON stdout for a structured outputs map and emit Dagster AssetMaterializations so they appear in the asset catalog and lineage views. (See Dagster’s docs on ops, graphs, job execution, and concurrency pools for operational polish.)
• Containers.
  CWL tools typically declare Docker images; cwltool will pull and run them if Docker/Podman is available, so your Dagster worker stays thin.
• IO managers.
  If you want stronger type/IO guarantees, it would need a small IOManager that maps output names to filesystem/S3 locations, then have downstream ops consume those paths.
• Versioning and lineage.
  Record the CWL file hash, image digests, and exact runner invocation in AssetMaterialization.metadata to make your runs reproducible and auditable.

[davidelbert]

The shortcoming of the simple pattern above is that the minimal run_cwl_op is a pretty monolithic step version. An advance would be to still keep it simple, but make it asset-aware. In other words, keep one Dagster op that runs the whole CWL, but surfaces CWL outputs and provenance as first-class Dagster metadata. To do that probably means additions to run_cwl_op:

• Structured outputs: parse cwltool JSON and return a typed Python dict. Emit AssetMaterialization per file or directory. Record CWL file hash, container image digests, exact runner args in metadata.
• Resources: put cwl_args and storage credentials into a Dagster ResourceDefinition so ops can share runner configuration cleanly.
• IOManager (optional?): map named CWL outputs to well-defined locations or object storage prefixes, so downstream assets can consume them without guessing paths.

This isn't fully granular but for almost no extra code you get Dagster's lineage, partitions, retries, schedules, sensors, and asset catalog while CWL handles the compute semantics.

[davidelbert]

Made a working proof of concept. Used @jan-janssen's simple arithmetic CWL from the PWD work as an example. Public repo is here: https://github.com/htmdec/dagster-cwl