Brain isn’t quite working today. I have a flow Parameter called

cluster

which determines whether a cluster is used which is accessed from a callback to

DaskCluster

. However, it’s not actually used in the flow which is causing flow run failures locally because it’s an ‘unexpected parameter’. I am 90% sure I’ve seen a way before where I can say to Prefect to relax and that it’s fine (technical terminology), but can’t remember exactly how

I think what you mean is adding a Parameter explicitly to the flow object using

.add_task()

?

from prefect import Flow
from prefect.executors import DaskExecutor

def dynamic_executor():
    from distributed import LocalCluster
    # could be instead some other class e.g. from dask_cloudprovider.aws import FargateCluster
    return LocalCluster(n_workers=prefect.context.parameters["n_workers"])

with Flow("example", executor=DaskExecutor(cluster_class=dynamic_executor)) as flow:
    flow.add_task(Parameter("n_workers", default=5))

Gotcha, thanks 🙂