Domino Flows

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

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 workflowfrom 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,)
@workflowdef 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.pyimport 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 stategit add -A && git commit -m "..." && git push
# Trigger remotelyPYTHONPATH=/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/