Quickstart

This page walks through a minimal pipeline — a single training task — to introduce the full pattern without getting distracted by multi-step complexity. For a pipeline with multiple tasks and more config structure, see Full Example: Training Pipeline.

workflow.py

import logging

import hydra
from flytekit import Resources, workflow
from flytekitplugins.omegaconf import OmegaConfTransformerMode, set_transformer_mode
from hydra.conf import HydraConf
from hydra.types import RunMode
from hydra_zen import builds, make_config, ZenStore
from launcher_conf import launcher_store
from omegaconf import DictConfig, MISSING

from scaffold.flyte import run_local_workflow, runtime_task, zen_call

logger = logging.getLogger(__name__)

set_transformer_mode(mode=OmegaConfTransformerMode.DictConfig)

workflow_store = ZenStore(name="workflow")


# ── 1. Configure Logic ────────────────────────────────────────────────────────


class Model:
    def __init__(self, name: str, learning_rate: float = 1e-3):
        self.name = name
        self.learning_rate = learning_rate


ModelConf = builds(Model, name=MISSING, populate_full_signature=True)

model_store = workflow_store(group="model")
model_store(ModelConf(name="resnet18", learning_rate=1e-3), name="resnet18")
model_store(ModelConf(name="resnet50", learning_rate=5e-4), name="resnet50")


# ── 2. Implement Logic ────────────────────────────────────────────────────────


def train(model: Model, data_path: str, output_path: str) -> str:
    logger.info(f"Training {model.name} on {data_path}, saving to {output_path}")
    return output_path


# ── 3. Configure Task and Workflow ────────────────────────────────────────────


TrainConf = builds(train, model=MISSING, data_path=MISSING, output_path=MISSING, populate_full_signature=True)

WorkflowConf = make_config(
    hydra_defaults=[
        "_self_",
        {"model@train.model": "resnet18"},
        {"override hydra/launcher": "flyte"},
    ],
    train=TrainConf,
)

workflow_store(
    make_config(
        bases=(WorkflowConf,),
        train=TrainConf(data_path="gs://example/data", output_path="gs://example/models"),
    ),
    name="default",
)


# ── 4. Implement Task and Workflow ────────────────────────────────────────────


@runtime_task(requests=Resources(mem="4Gi", cpu="2"))
def train_task(cfg: DictConfig, runtime_cfg: DictConfig) -> str:
    return zen_call(train, cfg)


@workflow
def training_pipeline(train: DictConfig, runtime_cfg: DictConfig) -> str:
    return train_task(cfg=train, runtime_cfg=runtime_cfg)


# ── 5. Run ────────────────────────────────────────────────────────────────────


@hydra.main(config_name="default", config_path=None, version_base="1.3")
def main(cfg: DictConfig) -> None:
    run_local_workflow(training_pipeline, cfg)


if __name__ == "__main__":
    workflow_store(HydraConf(mode=RunMode.MULTIRUN))
    workflow_store.add_to_hydra_store(overwrite_ok=True)
    launcher_store.add_to_hydra_store(overwrite_ok=True)
    main()

Running the pipeline

Locally:

python workflow.py hydra.launcher.execution_environment=local

This executes main() directly, which calls run_local_workflow. Flyte tasks and the workflow run as ordinary Python functions in the current process — no containers, no Flyte backend.

Remotely:

See the Deployment Configuration section for instructions on setting up your Flyte environment and configuring the launcher.

Overriding config from the CLI

Because the config is registered with Hydra, all standard Hydra overrides work:

# Use resnet50 instead of the default resnet18
python workflow.py train.model=resnet50

# Override a nested field
python workflow.py train.model.learning_rate=1e-4

# Show the rendered config without running
python workflow.py --cfg job

Pattern walkthrough

Step 1 — Configure Logic

builds() creates a structured config dataclass from a regular class. With populate_full_signature=True the config inherits the full signature including type hints, which Hydra uses for validation. MISSING marks required fields that have no default.

Multiple named variants are registered in a config group using a sub-store:

model_store = workflow_store(group="model")
model_store(ModelConf(name="resnet18"), name="resnet18")
model_store(ModelConf(name="resnet50"), name="resnet50")

The hydra_defaults list in WorkflowConf selects resnet18 by default and places it at train.model in the rendered config.

Step 2 — Implement Logic

Plain Python. No Flyte, no hydra, no Scaffold. This is intentional — functions are fully testable in isolation and zen_call handles the bridge to the config system.

Step 3 — Configure Task and Workflow

builds(train, ...) creates a config for the task. The named config workflow_store(..., name="default") provides the concrete values (paths, etc.) that fill in the MISSING fields.

Step 4 — Implement Task and Workflow

@runtime_task is a drop-in for @task that additionally:

• Calls apply_runtime_cfg(runtime_cfg) at the start of every remote execution to restore logging and other runtime settings inside the container

• Automatically excludes runtime_cfg from Flyte’s cache key, so changing log verbosity never invalidates a cached task result

zen_call(fn, cfg) instantiates all config fields (calling instantiate() on any builds() sub-config) and passes the results as keyword arguments to fn. In this example, cfg.model is a ModelConf — zen_call turns it into a Model instance.

The @workflow receives one DictConfig per task plus runtime_cfg, all resolved from the top-level config by run_local_workflow (locally) or by the launcher (remotely).

Step 5 — Run

run_local_workflow maps config keys to workflow inputs by name and executes the workflow. The __main__ block adds the HydraConf(mode=RunMode.MULTIRUN) entry to the store and flushes both stores before calling main().

Note

The launcher_conf.py file is imported at the top of the workflow file. It defines the project-specific Docker image configuration and is kept separate so it can be shared across multiple workflow files. See Deployment Configuration for its contents and all available options.