Retrievable transaction results #17545
-
|
Once a transaction is committed with persisted results, a file is written to disk with its name being the idempotency key if supplied. I'd like to know if this file could include a result attributed to the transaction so that it can be retrieved if need be in subsequent flow runs. Here is a minimal example which shares with my real use case some of the traits that prompted me to look for this. import time
from prefect import task, flow
@task
def task1():
time.sleep(2)
return "task1"
@task
def task2():
time.sleep(3)
return "task2"
@task
def task3(arg1):
time.sleep(1)
return "task3"
@task
def task4(arg1, arg2):
time.sleep(4)
return "task4"
@flow(persist_result=True)
def pipeline():
task1_future = task1.submit()
task2_future = task2.submit()
task3_future = task3.submit(task1_future)
task4_future = task4.submit(task2_future, task3_future)
result = task4_future.result()
return result
if __name__ == "__main__":
pipeline()In this workflow, I could enable caching for both tasks. But intermediate results may be objects that are tedious to serialise. Plus it would lead to needless disk use as I'm always only interested in the result of What I'd want is to be able to set a result for the transaction which could outlive the flow run: with transaction(key="my-key") as txn:
if txn.is_committed():
result = txn.get_result() # or txn.get("result") or something else along those lines
else:
task3_future = task3.submit(task1_future)
task4_future = task4.submit(task2_future, task3_future)
result = task4_future.result()
txn.set("result", result) # Or Prefect could commit the result automatically, e.g. if it's the return statement of the context manager block?Actually I thought of a workaround, but it feels like cheating: enable caching for @task(cache_key_fn: lambda context, parameters: "my-key")
def task4(arg1, cache_key):
time.sleep(4)
return "task4"
...
with transaction(key="my-key") as txn:
if txn.is_committed():
result = task4(task2_future, None) # Dummy input for task3
else:
task3_future = task3.submit(task1_future)
task4_future = task4.submit(task2_future, task3_future)
result = task4_future.result() |
Beta Was this translation helpful? Give feedback.
Replies: 2 comments 8 replies
-
|
Hi @TheoMathurin - I think if txn.is_committed():
result = txn.read().result |
Beta Was this translation helpful? Give feedback.
-
|
Oh wow, thanks @cicdw! I'll use this. How is this result determined? Is it the return statement of the transaction block? |
Beta Was this translation helpful? Give feedback.
-
|
Ah good question - so you would need to explicitly write/stage something to the with transaction(key="my-key") as txn:
if txn.is_committed():
result = txn.read().result # will be "foobar" on a second run
else:
txn.stage(value="foobar") # will be persisted at transaction commitUltimately all of this transactions machinery is what backs Prefect's task result and cache storage mechanism, so it's possible you could get what you need using tasks directly. |
Beta Was this translation helpful? Give feedback.
-
|
Thank you. I tried this and got an error on I also tried with staging bytes instead of a string, to no avail. |
Beta Was this translation helpful? Give feedback.
-
|
Can you share a fully reproducible example of what you mean? I think the best bet is to always use the |
Beta Was this translation helpful? Give feedback.
-
|
Sure! On second execution, the following script generates the aforementioned error for me: import os
from prefect import flow, task
from prefect.transactions import transaction
@task
def task1():
return "foobar"
@flow(persist_result=True, log_prints=True)
def my_flow():
with transaction(key="my-key") as txn:
# is.committed() does not work, see https://github.com/PrefectHQ/prefect/pull/17541
# if txn.is_committed():
if os.path.exists(
os.path.join(os.getenv("HOME"), ".prefect", "storage", "my-key")
):
print("Retrieving result")
value = txn.read().result
else:
print("Staging result")
value = task1()
txn.stage(value=value)
return value
my_flow() |
Beta Was this translation helpful? Give feedback.
-
|
NB: When I try to directly retrieve results from storage using the store = get_result_store()
result = store.read("my-key") # fails |
Beta Was this translation helpful? Give feedback.
Ah good question - so you would need to explicitly write/stage something to the
keyset on thetransactionusing thestagemethod:Ultimately all of this transactions machinery is what backs Prefect's task result and cache storage mechanism, so it's possible you could get what you need using tasks directly.