Jon11/28/2022, 6:19 PM
to get the results of a task called within a
. Seems the resource manager cleans up the task's result so it is unavailable to another flow?
└── 12:43:06 | ERROR | Task 'get_task_run_result': Exception encountered during task execution! Traceback (most recent call last): File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/engine/task_runner.py", line 880, in get_task_run_state value = prefect.utilities.executors.run_task_with_timeout( File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/utilities/executors.py", line 468, in run_task_with_timeout return task.run(*args, **kwargs) # type: ignore File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/tasks/prefect/flow_run.py", line 239, in get_task_run_result return task_run.get_result() File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/backend/task_run.py", line 79, in get_result self._result = self._load_result() File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/backend/task_run.py", line 85, in _load_result self._load_child_results() File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/backend/task_run.py", line 139, in _load_child_results task_run._assert_result_type_is_okay() File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/backend/task_run.py", line 162, in _assert_result_type_is_okay raise ValueError( ValueError: The task result has no `location` so the result cannot be loaded. This often means that your task result has not been configured or has been configured incorrectly.
Mason Menges11/28/2022, 8:46 PM
Jon11/28/2022, 8:50 PM
@prefect.task( name=HOIST_TASK_RESULT_TASK_NAME, result=results.LocalResult(), ) def hoist_task_result(task_result: any) -> any: """Used in conjunction with a Prefect resource_manager. Hoists a task's result out of a resource_manager's scope into the Flow's scope so the result can be accessed externally. Args: task_result (any): a Prefect tasks's result. Returns: any: the Prefect tasks's result. """ return task_result
with Flow() as flow: with ResourceManager() as rm: some_result = do_something() result_we_want_in_parent_flow = hoist_task_rersult(some_result)
with Flow() as flow: some_var = None with RM() as rm: some_var = do_something()
Mason Menges11/28/2022, 9:02 PM
I think the issue here ultimately boils down to the difference between registration/build time vs run time, basically when the flow is registered everything in the flow run context is statically defined so the "values" of the variables you have set aren't updated when the flow actually runs, storing it as a script may help this but I haven't tried this particular pattern in prefect 1. That said this is much more easily accomplished in prefect 2 which has much better support for native python patterns.
Jon11/29/2022, 11:31 AM
in a parent flow to get a task result in a child flow. when i put that task inside of a
, the result is suddenly unavailable with the error i shared above. im not sure how that interacts with build vs run time, since i am able to get the value that's generated during runtime when outside of the resource manager? it seems to me the issue here is that the resource manager is destroying the result?