0

我正在尝试根据列表强制定义任务。挑战在于该列表应基于 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()
4

1 回答 1

2

尝试在完美流程中动态创建任务最好通过映射功能进行管理。

但是，映射只会在流程运行期间从可迭代对象生成任务。它不会随意调整生成任务的依赖关系；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)
于 2020-11-02T04:06:18.083 回答