In Prefect core is there a simple way to invalidate cache if input of upstream tasks change?

user_2032998 0
Viewed 183

I'm trying to build a mechanism to invalidate cache for a task if upstream task constants input changes.

I currently make a hash of the input constants for each task, then use that as the target. I was thinking I could instead build up a hash of hashes of the current task inputs as well as the inputs of upstream tasks. Was thinking to do this by using constants and edges from the Flow object.

It seems like quite a common problem, was looking to see if there was in fact an easy way to achieve it?

prefect python-3.x python
1 Answers

In the end, I adapted my existing solution which uses xxhash to make hashes of the constants for each task.

My target for each task looked something like this before making the change:

'{task_name}_{hash_of_input_constants}'

The solution is in fact quite simple, I simply append the hash of input constants for the directly connected upstream task(s), so when there is only one directly connected upstream task target will look like this instead:

'{task_name}_{hash_of_input_constants_curent_task}_{hash_of_input_constants_directly_connected_upstream}'

This will invalidate the cache in the event there are changes to inputs upstream.

user_20329980
Related