flyte-migrate-control-flow

Flyte 1 to 2 Migration: Control Flow and Parallelism

Flyte 1 expressed branching, dynamic fan-out, and failure handling through DSL constructs (conditional(), @dynamic, @workflow(on_failure=...)) and map_task. In Flyte 2 these are all ordinary Python, because orchestration runs as real Python at runtime. Native if/elif/else replaces the conditional DSL, plain task loops replace @dynamic, try/except replaces on_failure, and flyte.map / asyncio.gather replace map_task.

Grounding References

Resource	URL
Migration guide (Control flow)	https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/control-flow/
Migration guide (Parallelism)	https://www.union.ai/docs/v2/flyte/user-guide/migration/flyte-2/parallelism/
Official docs	https://www.union.ai/docs/v2/flyte
Docs index (LLMs)	https://www.union.ai/docs/v2/flyte/llms.txt
SDK API reference	https://www.union.ai/docs/v2/union/api-reference/flyte-sdk/
Example code	https://github.com/unionai/unionai-examples
Flyte MCP tools	Available via flyte-mcp server

Conditional Execution

The conditional() DSL becomes ordinary Python if / elif / else — for example, choosing a model based on dataset size.

Flyte 1

from flytekit import task, workflow, conditional

@task
def train_gradient_boosting(n_rows: int) -> str:
    return f"trained gradient boosting on {n_rows} rows"

@task
def train_logistic_regression(n_rows: int) -> str:
    return f"trained logistic regression on {n_rows} rows"

@workflow
def main(n_rows: int) -> str:
    # Pick the model based on dataset size.
    return (
        conditional("model_choice")
        .if_(n_rows > 10_000)
        .then(train_gradient_boosting(n_rows=n_rows))
        .else_()
        .then(train_logistic_regression(n_rows=n_rows))
    )

Flyte 2

import flyte

env = flyte.TaskEnvironment(name="conditional")

@env.task
def train_gradient_boosting(n_rows: int) -> str:
    return f"trained gradient boosting on {n_rows} rows"

@env.task
def train_logistic_regression(n_rows: int) -> str:
    return f"trained logistic regression on {n_rows} rows"

# Branching is now ordinary Python control flow -- no conditional() DSL.
@env.task
def main(n_rows: int) -> str:
    if n_rows > 10_000:
        return train_gradient_boosting(n_rows)
    return train_logistic_regression(n_rows)