有没有办法在每个计划的流程运行时更新参数？

【问题标题】：Is there a way to update parameters as each scheduled flow run?有没有办法在每个计划的流程运行时更新参数？
【发布时间】：2020-11-13 11:16:57
【问题描述】：

我正在尝试找到一种方法来更新开始用于调度流程的每次迭代的参数。

例如，假设我有一个计划在一年的每个星期一运行一次的流程。对于第一个星期一，流程需要以 5 的参数运行。下星期一需要以 7 的参数运行，以此类推。每周运行所需的参数将更改一个常数。

根据文档，我似乎可以为每次运行创建一个带有相应参数的时钟，但对于为多次运行安排的流程来说，这似乎太过分了。

在 Prefect 中有更简单的方法吗？

【问题讨论】：

在问题中添加一些代码会增加获得答案的机会：)

标签： prefect

【解决方案1】：

这听起来像是您想将某种形式的业务逻辑封装到您的 Parameter 计算中，这是向您的 Flow 添加新的 Task 的一个很好的用例：

import prefect
from prefect import task, Flow, Parameter


varying_param = Parameter("param", default=None) # none as the default

@task(name="Varying Parameter")
def param_calculation(p):
    time = prefect.context.scheduled_start_time 

    # if a value was provided, use it
    if p is not None:
        return p
    
    # do some calculations to decide what value is appropriate
    # and return it


with Flow("Minimal example") as flow:
    param_value = param_calculation(varying_param)
    # now use this value in downstream tasks

【讨论】：