Cache across task/flow runs in different process with dask

[nicolls1]

Current behavior

The docs here: https://docs.prefect.io/core/examples/cached_task.html document how to cache in a single flow. I am wondering if its possible to extend this to flows run from multiple processes.

Here in Output Caching (https://docs.prefect.io/core/concepts/persistence.html#output-caching) it mentions:

The cache is stored in context

Note that when running Prefect Core locally, your Tasks' cached states will be stored in memory within prefect.context.

I assume the dask executor still uses the prefect core locally?

Proposed behavior

Store cached outputs in dask with a distributed variable (https://distributed.dask.org/en/latest/api.html?highlight=queue#distributed.Variable) so that when you run the example script below 2 times the cached value is used.

Example

This modified docs example doesn't return the same random number when run 2 times in a row

import datetime
import random

from prefect import task, Flow
from prefect.engine.executors import DaskExecutor


@task(cache_for=datetime.timedelta(seconds=30), cache_key='key')
def return_random_number():
    return random.random()


@task
def print_number(num):
    print("=" * 50)
    print("Value: {}".format(num))
    print("=" * 50)


with Flow("cached-task") as flow:
    result = print_number(return_random_number)


executor = DaskExecutor(address="tcp://localhost:8786")
flow.run(executor=executor)

Maybe this is possible at the moment but I'm not sure how I could do it from the docs. Would be happy to document if that is the case.

[joshmeek]

Hey @nicolls1 instead of using the cache_for and cache_key I think you should use the newer Results / Targets interface to cache the results of your local flow runs somewhere like on disk https://docs.prefect.io/core/concepts/persistence.html#output-caching-based-on-a-file-target

[jcrist]

Yeah, using the Result interface will be much more robust than trying to hack something together to store things in the dask cluster. #2619 about integrating results with cache_for/cache_validator may also interest you.

[nicolls1]

Thanks for the feedback! Looking into the result interface now and I'll see if I can make a new class that will keep a task in dask's distributed memory.

[github-actions]

This issue is stale because it has been open 30 days with no activity. To keep this issue open remove stale label or comment.

[github-actions]

This issue was closed because it has been stale for 14 days with no activity. If this issue is important or you have more to add feel free to re-open it.