Advanced Patterns

Dynamic workflows

A dynamic workflow lets you write conditional or data-dependent control flow that Flyte evaluates at runtime rather than at compile time.

Note

Prefer static workflows when possible — dynamic workflows only fail at runtime, making errors harder to catch early. For running the same function many times with different inputs, a map task is usually the better choice.

Dynamic workflows are opaque to Flyte’s task discovery. The launcher cannot automatically detect tasks nested inside a @dynamic function. Declare them explicitly with @mxm_register so the launcher can find and register them:

from flytekit import dynamic, workflow
from scaffold.flyte.core import mxm_register

@mxm_register(nodes=[my_task])
@dynamic
def dynamic_pipeline(cfg: DictConfig, runtime_cfg: DictConfig) -> None:
    my_task(cfg=cfg, runtime_cfg=runtime_cfg)

@workflow
def pipeline(cfg: DictConfig, runtime_cfg: DictConfig) -> None:
    dynamic_pipeline(cfg=cfg, runtime_cfg=runtime_cfg)

Note that a static @workflow is still required as the entry point — the launcher always needs a top-level workflow to identify.

Cron Jobs

Workflows can run on a cron schedule by setting cron_schedule on FlyteWorkflowConf:

# launcher_conf.py
launcher_store(
    FlyteLauncherConf(
        workflow=FlyteWorkflowConf(
            ...,
            cron_schedule="0 5 * * *",  # every day at 5am
        )
    ),
    name="flyte",
    group="hydra/launcher",
)

The recommended pattern is to leave cron_schedule unset in the config file and only apply it in CI/CD for staging/production registrations, so development runs are never accidentally scheduled:

# In your CI/CD pipeline (e.g. Cloud Build), targeting the staging domain:
python workflow.py hydra.launcher.workflow.cron_schedule="0 5 * * *" \
                   hydra.launcher.workflow.domain=staging

Managing a schedule after registration:

To deactivate an active launchplan (after port-forwarding flyteadmin 30081:81):

flytectl update launchplan -p default -d development \
    --version <version> <launchplan-name> --archive

Re-activate with --activate. The launchplan name is logged at the end of registration and shown in the Flyte UI under Launch Workflow.

Note

Flyte requires workflows with a cron schedule to accept a kickoff_time argument:

from datetime import datetime

@workflow
def pipeline(cfg: DictConfig, runtime_cfg: DictConfig, kickoff_time: datetime = datetime.now()) -> ...:
    ...

Multi-input workflows

Some workflows accept additional inputs beyond the config — for example, a data path or a request payload passed at execution time rather than baked into the config. The workflow interface supports this naturally:

@workflow
def pipeline(cfg: DictConfig, data_string: str, runtime_cfg: DictConfig) -> str:
    ...

run_local_workflow maps config keys to workflow inputs by name. Any workflow parameter not found in the config is left to the workflow’s default value (a warning is logged) and can be passed when calling the workflow function. For local execution, pass the extra arguments directly to run_local_workflow:

run_local_workflow(pipeline, cfg, data_string="my data")

For remote execution, we need to configure the launcher to not run the workflow directly after registration (as there we want to provide the extra arguments at execution time):

# launcher_conf.py
launcher_store(
    FlyteLauncherConf(
        ...,
        run=False,  # Don't execute immediately after registration
    ),
    name="flyte",
    group="hydra/launcher",
)

Then we can execute the workflow with the extra arguments using FlyteRemote:

from scaffold.flyte.flyte_utils import FlyteRemoteHelper

LAUNCHPLAN = "hydra_workflow_cfg_workflow_with_extra_inputs_default_0"
ADMIN_ENDPOINT = "localhost:30081"
DOMAIN = "development"

dummy_string = "my data"

# execute the workflow that was registered by flyte launcher before
FlyteRemoteHelper(domain=DOMAIN, admin_endpoint=ADMIN_ENDPOINT).execute_flyte_launchplan(
    LAUNCHPLAN, {"data_string": dummy_string}

)