python - Prefect:如何基于从参数派生的任务列表强制创建任务
问题描述
我正在尝试根据列表强制定义任务。挑战在于该列表应基于 Prefect 参数。
下面是我尝试过的代码,但显然它不起作用,因为 task_dependency_pairs 是一个任务,而不是一个列表。
如何在不破坏 Parameter 任务和其他动态生成的任务之间的依赖关系的情况下使其工作?
from prefect import task, Flow, Parameter, Task
import time
@task
def task_dependency_pairs(param):
return [
('task 1', f'{param}A', ''),
('task 2', f'{param}B','task 1'),
('task 3', f'{param}C','task 1')]
class Task_class(Task):
def run(self, **kwarg):
time.sleep(5)
print(f"This task {kwarg['task_name']} does a lot of things with {kwarg.get('calc_value','')}.")
for task_name, dependency in task_dependency_pairs:
globals()[task_name] = type(task_name, (Task_class,),{"__module__": __name__})
with Flow("my_process") as flow:
param = Parameter("param", default="default_param")
task_dependency_pairs_list = task_dependency_pairs(param)
for task_name, calc_value, dependency in task_dependency_pairs_list: # This won't work
task_instance = globals()[task_name](name=task_name)
flow.add_task(task_instance(task_name = task_name, calc_value = calc_value))
for task_name, calc_value, dependency in task_dependency_pairs_list: # This won't work
if len(dependency) >0:
flow.get_tasks(name=task_name)[0].set_upstream(flow.get_tasks(name=dependency)[0])
flow.visualize()
解决方案
尝试在完美流程中动态创建任务最好通过映射功能进行管理。
但是,映射只会在流程运行期间从可迭代对象生成任务。它不会随意调整生成任务的依赖关系;mapped
它们都共享主要任务中定义的依赖关系。
但是,如果您想在运行时生成流(具有编程依赖项),我能想到的唯一方法是创建一个创建流并立即运行它的任务。
这可能会寻找您的流程的方式是:
...
@task
def run_flow(inputs):
with Flow("subflow") as sub_flow:
for (name, calc_value, dependency) in inputs:
inst = Task_class(name=name)(task_name=name, calc_value=calc_value)
sub_flow.add_task(inst)
if dependency:
inst.set_upstream(sub_flow.get_tasks(name=dependency)[0])
sub_flow.run()
with Flow("my_process") as flow:
param = Parameter("param", default="default_param")
task_dependency_pairs_list = task_dependency_pairs(param)
run_flow(task_dependency_pairs_list)
推荐阅读
- node.js - 使用角度和节点上传图像的问题
- c - union 对变量有什么帮助?
- python - Visual Studio for Python 项目中的 Flask 导入问题
- rust - Into 的多个泛型
- swift - RealityKit AR 方向
- html - Angular 再次加载预加载的 svg
- sql - PostgreSQL 查询获取 12 个月内出现多次的值
- compression - FitGirl 是如何做到如此高的压缩比的呢?
- javascript - Firebase rtdb 无法在构造函数中绑定值
- mysql - 在创建表时将默认值作为 Mysql 中两列的差异