Flyte使用map_task无法实现任务并行运行求助

先修正代码基础错误 #

你的代码里用到了time.sleep但未导入time模块，先补上这一行：

import time
from flytekit import map_task, task, workflow

@task
def do_something(value: str) -> str:
    print(f"launched: {value}", flush=True)
    time.sleep(60)  # 模拟耗时操作
    return f"{value}-processed"


@workflow
def do_multiple_things() -> list[str]:
    values = ["foo", "bar", "baz"]
    # 指定并发数，确保集群执行时的并行度
    return map_task(do_something, concurrency=3)(value=values)

本地运行为何是串行？ #

用flytekit run在本地执行时，Flyte默认以单进程串行方式运行map_task，这是本地调试的默认行为，方便排查问题。

如果想在本地模拟并行，可通过环境变量指定并行数：

FLYTE_MAP_PARALLELISM=3 flytekit run your_script.py do_multiple_things

真正实现并行的方式 #

要让任务真正并行执行，需将工作流提交到Flyte集群运行。集群环境中，map_task会根据concurrency参数（或集群默认配置）启动多个独立任务实例，每个实例在单独计算节点上并行处理——此时三个do_something任务会同时启动，各自休眠60秒后完成。

验证并行的方法 #

在集群上运行时，通过Flyte控制台可看到三个子任务同时处于「Running」状态，而非依次执行。

内容的提问来源于stack exchange，提问作者duncan