Migration overview

This section walks through the Flyte 1 workload patterns you already know — data ETL, model training, hyperparameter sweeps, batch inference — and their Flyte 2 equivalents. Every pattern is a complete, runnable v1↔v2 example pair in the unionai-examples repository.

Two conceptual shifts motivate almost every change — pure Python execution and the asynchronous model — after which most migrations come down to a couple of mechanical moves. This page covers both shifts, the terminology mapping, and a quick-reference module.

Pure Python execution

In Flyte 1, @workflow functions were constrained to a DSL subset of Python that compiled to a static DAG. In Flyte 2 there is no @workflow decorator: everything is a @env.task, and a “workflow” is simply a task that calls other tasks. Orchestration runs as real Python at runtime, so loops, conditionals, and try/except work anywhere.

import flytekit

image = flytekit.ImageSpec(
    name="hello-world-image",
    packages=["requests"],
)

@flytekit.task(container_image=image)
def mean(data: list[float]) -> float:
    return sum(list) / len(list)

@flytekit.workflow
def main(data: list[float]) -> float:
    output = mean(data)

    # ❌ performing trivial operations in a workflow is not allowed
    # output = output / 100

    # ❌ if/else is not allowed
    # if output < 0:
    #     raise ValueError("Output cannot be negative")

    return output

import flyte

env = flyte.TaskEnvironment(
    "hello_world",
    image=flyte.Image.from_debian_base().with_pip_packages("requests"),
)

@env.task
def mean(data: list[float]) -> float:
    return sum(data) / len(data)

@env.task
def main(data: list[float]) -> float:
    output = mean(data)

    # ✅ performing trivial operations in a workflow is allowed
    output = output / 100

    # ✅ if/else is allowed
    if output < 0:
        raise ValueError("Output cannot be negative")

    return output

This unlocks workflows that adapt to runtime conditions, native try/except error handling, and the intuitive composition of ordinary Python functions.

Asynchronous model

Flyte 2 is built on Python’s asyncio, with a crucial twist: the Flyte orchestrator acts as the event loop, scheduling awaited tasks across distributed infrastructure. This makes async/await the natural way to express parallelism.

The core async keywords carry Flyte-specific meaning:

• async def declares a coroutine.
• await signals where a task can be scheduled in parallel — not just an I/O yield point.
• asyncio.gather tells the orchestrator that a set of tasks are independent and can be distributed across separate compute resources.

Consider this pattern for parallel data processing:

import asyncio
import flyte

env = flyte.TaskEnvironment("data_pipeline")

@env.task
async def process_chunk(chunk_id: int, data: str) -> str:
    # This could be any computational work - CPU or I/O bound
    await asyncio.sleep(1)  # Simulating work
    return f"Processed chunk {chunk_id}: {data}"

@env.task
async def parallel_pipeline(data_chunks: list[str]) -> list[str]:
    # Create coroutines for all chunks
    tasks = []
    for i, chunk in enumerate(data_chunks):
        tasks.append(process_chunk(i, chunk))

    # Execute all chunks in parallel
    results = await asyncio.gather(*tasks)
    return results

In standard Python this mainly benefits I/O-bound work; in Flyte 2 the orchestrator schedules each process_chunk task on its own pod.

True parallelism for all workloads

Async syntax in Flyte 2 is not just for I/O-bound operations. When the orchestrator encounters await asyncio.gather(...), it runs those independent tasks simultaneously across compute resources — achieving true parallelism for CPU-bound work (model training, heavy math), I/O-bound work (queries, API calls), and mixed workloads alike.

Calling sync tasks from async tasks

You don’t need to rewrite existing synchronous code. Flyte automatically “asyncifies” sync functions; just call them from an async context with .aio():

@env.task
def legacy_computation(x: int) -> int:
    # Existing synchronous function works unchanged
    return x * x + 2 * x + 1

@env.task
async def modern_workflow(numbers: list[int]) -> list[int]:
    # Call sync tasks from async context using .aio()
    tasks = []
    for num in numbers:
        tasks.append(legacy_computation.aio(num))

    results = await asyncio.gather(*tasks)
    return results

The flyte.map function: Familiar patterns

For code that used Flyte 1’s map, flyte.map is a direct replacement that works in both sync and async contexts:

@env.task
def sync_map_example(n: int) -> list[str]:
    # Synchronous version for easier migration
    results = []
    for result in flyte.map(process_item, range(n)):
        if isinstance(result, Exception):
            raise result
        results.append(result)
    return results

@env.task
async def async_map_example(n: int) -> list[str]:
    # Async version using flyte.map - exact pattern from SDK examples
    results = []
    async for result in flyte.map.aio(process_item, range(n), return_exceptions=True):
        if isinstance(result, Exception):
            raise result
        results.append(result)
    return results

It provides dual interfaces (flyte.map.aio() and flyte.map()), return_exceptions for graceful failure handling (a more flexible replacement for Flyte 1’s min_success_ratio), automatic UI grouping, and optional concurrency limits.

Terminology and concept mapping

Several Flyte 1 concepts were renamed or reshaped in Flyte 2. The table below maps the ones you’ll meet most often.

Package imports

The package is renamed from flytekit to flyte, and the workflow/dynamic/map_task imports disappear:

import flytekit
from flytekit import task, workflow, dynamic, map_task
from flytekit import ImageSpec, Resources, Secret
from flytekit import current_context, LaunchPlan, CronSchedule

import flyte
from flyte import TaskEnvironment, Resources, Secret
from flyte import Image, Trigger, Cron

The two mechanical changes behind (almost) every migration

Most of a migration comes down to two moves:

1. Move task configuration into a TaskEnvironment

Instead of configuring the image, resources, and caching on each task decorator, configure them once on a flyte.TaskEnvironment and share it across tasks:

env = flyte.TaskEnvironment(
    name="training",
    image=flyte.Image.from_debian_base().with_pip_packages("scikit-learn", "pandas"),
    resources=flyte.Resources(cpu="2", memory="4Gi"),
    cache="auto",
)

2. Replace @task / @workflow / @dynamic with @env.task

Every decorated function becomes an @env.task. There is no separate workflow or dynamic construct: a “workflow” is simply a task that calls other tasks, and orchestration is plain Python.

In this section

The migration patterns are grouped by theme. Start with Tasks and workflows, then jump to whatever your workload needs:

• Tasks and workflows — the structural shift: @task/@workflow → @env.task, sequential ordering, nested “subworkflows”, and the @task → TaskEnvironment parameter mapping.
• Task configuration — moving image/resources/cache to the TaskEnvironment, GPUs, secrets, caching, and scheduling with triggers.
• CLI and configuration — pyflyte → flyte command mapping and config-file changes.
• Control flow — conditional() and @dynamic become plain Python if/loops, and on_failure becomes try/except.
• Parallelism and fan-out — map_task → flyte.map / asyncio.gather, plus a data-backfill example.
• Data types and I/O — FlyteFile/FlyteDirectory → flyte.io.File/Dir, StructuredDataset → flyte.io.DataFrame, dataclasses, and an ETL example.
• ML workloads — small-model training, hyperparameter optimization, deep learning, batch inference, and an end-to-end pipeline.
• New in Flyte 2 — patterns that weren’t possible in Flyte 1 at all: real-time model serving, batch inference, apps, and sandboxed code execution.
  • Union features — migrating Union-specific Actors (→ reusable containers) and Apps (→ the Flyte SDK).
• Hybrid v1 and v2 pipelines — calling between v1 and v2 in both directions during the transition.
• Gotchas and caveats — common gotchas plus the deeper caveats of the new execution model, including non-deterministic behavior and keeping orchestration lightweight.

Quick reference

A minimal Flyte 2 module, end to end:

import asyncio
import flyte

# 1. Define an image
image = (
    flyte.Image.from_debian_base(python_version=(3, 11))
    .with_pip_packages("pandas", "numpy")
)

# 2. Create a TaskEnvironment
env = flyte.TaskEnvironment(
    name="my_env",
    image=image,
    resources=flyte.Resources(cpu="1", memory="2Gi"),
)

# 3. Define tasks
@env.task
async def process(x: int) -> int:
    return x * 2

# 4. Define the entrypoint task
@env.task
async def main(items: list[int]) -> list[int]:
    results = await asyncio.gather(*[process(x) for x in items])
    return list(results)

# 5. Run it
if __name__ == "__main__":
    flyte.init_from_config()
    run = flyte.run(main, items=[1, 2, 3, 4, 5])
    print(run.url)
    run.wait()

# CLI
flyte run my_module.py main --items '[1,2,3,4,5]'   # remote (default)
flyte run --local my_module.py main --items '[1,2,3,4,5]'
flyte deploy my_module.py my_env