Jason Bertman
10/06/2022, 6:40 PM@flow(
task_runner=RayTaskRunner(
address="<ray://ray-cluster-kuberay-head-svc:10001>"
),
)
def main(...):
...
if __name__ == "__main__": # If this flow is called directly, don't use the address
main(..., task_runner=RayTaskRunner)
Zanie
10/06/2022, 6:44 PMPREFECT__FLOW_RUN_ID
Jason Bertman
10/06/2022, 6:45 PMRyan Peden
10/06/2022, 6:45 PMif __name__ == "__main__":
flow_run = main.with_options(task_runner=RayTaskRunner())
flow_run()
Jason Bertman
10/06/2022, 6:45 PMwith_options
, will give that a tryZanie
10/06/2022, 6:47 PMif __name__ == "__prefect_loader__"
to detect a deployment (I’d double check the name but I think that’s it).Jason Bertman
10/06/2022, 6:47 PMZanie
10/06/2022, 6:48 PMdef get_task_runner():
if "PREFECT__FLOW_RUN_ID" in os.environ:
return RayTaskRunner(
address="<ray://ray-cluster-kuberay-head-svc:10001>"
)
else:
...
@flow(
task_runner=get_task_runner()
)
Jason Bertman
10/06/2022, 6:49 PM