Jon
11/28/2022, 6:19 PMget_task_run_result
to get the results of a task called within a resource_manager
. 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 Menges
11/28/2022, 8:46 PMJon
11/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 Menges
11/28/2022, 9:02 PMstored_as_script=True
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.Jon
11/29/2022, 11:31 AMget_task_run_result
in a parent flow to get a task result in a child flow. when i put that task inside of a resource_manager
, 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?