domino-flows

name	domino-flows
description	Orchestrate multi-step ML workflows using Domino Flows (built on Flyte). Define DAGs with typed inputs/outputs, heterogeneous environments, automatic lineage, and reproducibility. Use when building data pipelines, multi-stage training workflows, or processes requiring orchestration and monitoring.

Domino Flows Skill

This skill provides comprehensive knowledge for orchestrating ML workflows using Domino Flows, built on the Flyte platform.

Key Concepts

What are Domino Flows?

Domino Flows enable:

• DAG-based orchestration: Define workflows as directed acyclic graphs
• Typed interfaces: Strong typing for inputs and outputs
• Heterogeneous environments: Different environments per task
• Automatic lineage: Track data and model provenance
• Reproducibility: Version-controlled workflows
• Scalability: Distributed execution across compute resources

Core Components

Component	Description
Task	Single unit of work (runs as a Domino Job)
Workflow	DAG connecting tasks
Artifact	Typed input/output passed between tasks
Launch Plan	Configured workflow execution

Related Documentation

• FLOW-BASICS.md - DAG concepts, task definitions
• EXAMPLES.md - Common flow patterns

Quick Start

⚠️ Critical: Domino Flows does NOT support native Flyte @task decorators. Tasks must use DominoJobTask + DominoJobConfig. Only @workflow is unchanged.

Basic Flow

Each task runs as a Domino Job. Stage scripts read from /workflow/inputs/<name> and write to /workflow/outputs/o0. Pass PYTHONPATH=/mnt/code in the command.

from flytekit import workflow
from flytekitplugins.domino.task import DominoJobConfig, DominoJobTask

preprocess_task = DominoJobTask(
    name="Preprocess Data",
    domino_job_config=DominoJobConfig(
        Command="bash -c 'PYTHONPATH=/mnt/code python /mnt/code/stages/preprocess.py'",
    ),
    inputs={"input_path": str},
    outputs={"o0": str},
    use_latest=True,
)

train_task = DominoJobTask(
    name="Train Model",
    domino_job_config=DominoJobConfig(
        Command="bash -c 'PYTHONPATH=/mnt/code python /mnt/code/stages/train.py'",
    ),
    inputs={"preprocess_output": str},
    outputs={"o0": str},
    use_latest=True,
)

@workflow
def training_pipeline(input_path: str = "/mnt/data/raw.csv") -> str:
    preprocess_output = preprocess_task(input_path=input_path)
    result = train_task(preprocess_output=preprocess_output)
    return result

Stage Script Pattern

# stages/preprocess.py
import json, os

INPUTS, OUTPUTS = "/workflow/inputs", "/workflow/outputs"

def main():
    input_path = open(f"{INPUTS}/input_path").read().strip()
    # ... do work ...
    os.makedirs(OUTPUTS, exist_ok=True)
    with open(f"{OUTPUTS}/o0", "w") as f:
        f.write(json.dumps({"output_path": "/mnt/artifacts/processed.parquet"}))

if __name__ == "__main__":
    main()

Running the Flow

# Always commit and push first — jobs run against remote repo state
git add -A && git commit -m "..." && git push

# Trigger remotely
PYTHONPATH=/mnt/code pyflyte run --remote \
    my_flow.py training_pipeline \
    --input_path "/mnt/data/raw.csv"

When to Use Flows

Good Use Cases

• Data processing → Model training pipelines
• ETL with ML steps
• Multi-stage training with different environments
• Processes requiring reproducibility and lineage
• Scheduled/triggered workflows

Not Ideal For

• Single dataset with many small computations
• Tasks that write to mutable shared state
• Simple single-step processes
• Real-time inference (use Model APIs instead)

Documentation Links

• Domino Flows: https://docs.dominodatalab.com/en/latest/user_guide/78acf5/orchestrate-with-flows/
• Flyte Documentation: https://docs.flyte.org/