本文详解 flyte 中 `map_task` 的并行执行机制，明确区分本地调试与远程集群行为差异，并推荐使用最新实验性 `map_task`，附完整可运行示例与关键注意事项。
在 Flyte 中实现真正意义上的并行任务执行，关键在于理解其执行上下文与 API 演进。许多初学者会发现：本地运行（pyflyte run）时，即使使用 map_task，任务也表现为串行执行——这并非代码错误，而是当前 flytekit 本地执行器的设计限制。
✅ 正确的并行执行前提
Flyte 的并行能力仅在连接真实后端（如 FlyteAdmin + K8s 集群）时生效。map_task 的本质是将单个任务实例化为多个独立工作流节点（即“映射展开”），由 Flyte 控制平面统一调度至不同 Worker Pod 并发执行。而本地执行（--local 或默认 pyflyte run）仅模拟逻辑流程，所有子任务共享同一 Python 进程，无法突破 GIL 限制，因此必然串行。
⚠️ 注意：截至 2024 年中，flytekit 尚未支持本地多进程/多线程并行执行 map_task。该功能属于长期演进计划（FlyteKit Parallel Local Execution），暂无稳定发布日期。
✅ 推荐用法：采用实验性 map_task
官方已将更健壮、更符合云原生调度语义的新版 map_task 移入 flytekit.experimental 模块。它修复了旧版在类型推导、错误传播和资源隔离方面的若干问题，并作为未来默认实现的基础。
以下是修正后的完整工作流示例（兼容远程执行）：
import time
from flytekit import task, workflow
from flytekit.experimental import map_task # ✅ 使用 experimental 版本
@task
def do_something(value: str) -> str:
print(f"[{time.time():.0f}] START: {value}", flush=True)
time.sleep(5) # 模拟耗时操作（实际应为 I/O 或计算密集型）
print(f"[{time.time():.0f}] DONE: {value}", flush=True)
return f"{value}-processed"
@workflow
def do_multiple_things() -> list[str]:
values = ["foo", "bar", "baz"]
# map_task 自动展开为 3 个并行任务节点
return map_task(do_something)(value=values)部署到远程 Flyte 集群后，您将在 UI 中清晰看到三个并行运行的 do_something 节点，日志时间戳将高度重叠，证实真正的并发执行。
✅ 关键注意事项与最佳实践
- 输入必须为列表（List）且长度确定：map_task 要求 value= 参数传入 List[T]，不可为生成器或动态长度结构。
- 任务需无状态、彼此独立：map_task 不提供跨实例通信机制，严禁在 do_something 中读写共享文件或数据库状态。
-
资源声明建议显式化：为避免 K8s 调度争抢，应在 @task 中指定资源：
@task(requests=Resources(cpu="1", mem="512Mi")) def do_something(...): ...
- 错误处理：任一子任务失败将导致整个 map_task 失败。如需容错，请改用 dynamic workflow + asyncio.gather 或自定义重试策略。
- 调试技巧：本地开发时，可用 print + time.time() 粗略验证逻辑分片；最终并行性务必通过远程执行 + Flyte Console 日志/时间轴确认。
总结而言，Flyte 的并行能力是平台级特性，而非 SDK 单机模拟。拥抱 flytekit.experimental.map_task，部署至真实集群，即可释放其高并发调度潜力——这才是云原生工作流编排的正确打开方式。
