https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Can you explain what does this cache ...
# marvin-ai
l
@Marvin Can you explain what does this cache policy do and how can it affect flow's performance - @task(cache_policy=NO_CACHE)?
m
✅ thought for 182.0 seconds
Short answer: 
@task(cache_policy=NO_CACHE)
disables caching and result persistence for that task. Prefect won’t compute a cache key, won’t read any prior result, and won’t write the task’s result to storage—so the task function runs every time. What that means for performance - Slower if the task is expensive: because you cannot reuse a previous result, every run (and every repeated call within a run) recomputes the work. - Slightly faster per call if the task is cheap: you avoid cache-key computation and result persistence I/O. - No deduping under concurrency: if multiple runs/threads hit the same task simultaneously, they’ll all execute instead of one producing a result that others reuse. - No cache expiration/refresh effects: 
cache_expiration
and 
refresh_cache
don’t apply when using 
NO_CACHE
. When to use it - Non-deterministic or side-effecting tasks (e.g., calling external APIs, writing to DBs, using timestamps/randomness). - Debugging or when you explicitly want to force re-execution. - When result persistence is unnecessary overhead. Notes - This is task-scoped; other tasks in the flow are unaffected. - In Prefect 3.x, the default cache policy is a composition (
DEFAULT = INPUTS + TASK_SOURCE + RUN_ID
), which primarily helps within a single flow run and for concurrency control. 
NO_CACHE
disables even that behavior. - Setting 
persist_result=False
will also effectively disable caching for that task because there’s nothing to store or retrieve. Docs and reference - Caching concepts: Prefect 3.x caching concepts - How-to and patterns: Cache workflow steps - API reference: prefect.cache_policies (see 
NO_CACHE
) If you share what the task does and where you’re seeing bottlenecks, I can suggest a cache policy (or isolation level) that balances performance and correctness for your case.
l
@Marvin I want to start processing the results i get as soon as they arrive without waiting for all of them but here as_completed doesn't work, why:
Copy code
@flow(task_runner=ThreadPoolTaskRunner())
def timocom_search_flow(
    search_parameters: dict,
    pg_secret_block: str,
    timocom_secret_block: str,
    rabbitmq_secret_block: str,
    ors_server_variable: str,
    price_calculation_server_variable: str,
    predictive_scoring_server_variable: str,
    here_maps_api_key_variable: str,
):
    logger = get_run_logger()
    logger.warning(f"Starting Timocom search with parameters: {search_parameters} at {datetime.now()}")

    try:
        sanitized_search_parameters = sanitize_and_validate_search_parameters(search_parameters)
        pg_utils = build_pg_utils_from_block(pg_secret_block)

        # Perform Timocom search

        # Timocom credentials are stored as "tenant_id": {"user": "username", "pass": "password"}
        creds = Secret.load(timocom_secret_block).get()
        tenant_ref = sanitized_search_parameters["tenant_ref"]
        logger.warning(f"Using Timocom credentials for tenant: {tenant_ref}")
        tenant_creds = creds.get(tenant_ref)
        if not tenant_creds:
            raise ValueError(f"No Timocom credentials found for tenant {tenant_ref}")

        user = tenant_creds["user"]
        password = tenant_creds["pass"]

        # Initialize Timocom instance
        timocom_instance = Timocom(pg_utils, user, password)
        # Create search tasks
        search_tasks = timocom_instance.create_search_tasks(sanitized_search_parameters)
        search_futures = timocom_instance.search.map(search_tasks)

        logger.warning("Timocom search process completed.")

        config = get_config(
            ors_server_variable,
            price_calculation_server_variable,
            predictive_scoring_server_variable,
            here_maps_api_key_variable,
        )

        # for future in search_futures:
        for finished_future in as_completed(search_futures):
            res = finished_future.result()
            # res = future.result()
            if not res:
                logger.warning("No results found for this search task.")
                continue
            for batch in chunked(res, 100):
                logger.warning(f"Processing batch of {len(batch)} results starting at {datetime.now()}.")

                distance_mapping = get_distance_mapping(batch, config)
                proccessed_results = []
                for i in range(len(batch)):
                    proccessed_results.append(process_single(config, batch[i], distance_mapping[i]))
                logger.warning(f"Post-processing of all freight requests completed at {datetime.now()}.")

                save_requests_pg(proccessed_results, pg_secret_block)

                logger.warning("Requests data persistence completed.")
                logger.warning(f"Publish message: {datetime.now()}")
                publish_refresh_message(
                    rabbitmq_secret_block,
                    search_parameters["tab_id"],
                    search_parameters["job_id"],
                    "REFRESH",
                )

    except Exception as e:
        logger.error(f"Error during Timocom search initiation: {e}")
        publish_refresh_message(
            rabbitmq_secret_block, search_parameters["tab_id"], search_parameters["job_id"], "FAILED"
        )
        raise
m
✅ thought for 210.3 seconds
Great goal — Prefect can absolutely stream results as they finish. In Prefect 3.x you should use Prefect’s own 
as_completed
with Prefect futures from a mapped task. A few common reasons it “doesn’t work”: Most common pitfalls 1) Wrong import - If you imported 
as_completed
from 
concurrent.futures
, it won’t work with `PrefectFuture`s. - Fix:
Copy code
from prefect.futures import as_completed
Docs: prefect.futures API 2) Not actually mapping a Prefect task - 
timocom_instance.search
must be a Prefect task (i.e., created with 
@task
). If it’s a plain Python method, 
.map(...)
here isn’t Prefect mapping and won’t return `PrefectFuture`s. - Quick check:
Copy code
logger.warning(type(search_futures[0]))
  # should be <class 'prefect.futures.PrefectTaskRunFuture'> (or similar)
3) Heavy, blocking work in the flow thread - Your post-processing (chunking, distance mapping, DB writes, publish messages) runs in the flow thread. That’s okay, but if it’s very heavy it can make it seem like nothing is “arriving.” Tasks still run concurrently in background threads, but the flow only returns to the 
as_completed
loop after your processing finishes. - If that’s the case, consider moving the heavy steps into tasks as well so the flow thread stays responsive and you keep concurrency high. 4) Shared state/thread-safety - If 
Timocom
or any client is shared across threads, it may serialize access internally and you won’t see true concurrency. Prefer constructing the client inside a task per call, or otherwise ensure the client is thread-safe. A minimal working pattern
Copy code
from prefect import flow, task
from prefect.futures import as_completed
from prefect.task_runners import ThreadPoolTaskRunner

@task
def search_one(item):
    # do the work
    return item

@flow(task_runner=ThreadPoolTaskRunner())
def my_flow(items):
    futures = search_one.map(items)
    for fut in as_completed(futures):
        # optionally: fut.result(raise_on_failure=False) if you want to keep going on failures
        res = fut.result()
        # process this result immediately
        print("got:", res)
Apply that to your code - Ensure the import:
Copy code
from prefect.futures import as_completed
- Ensure 
timocom_instance.search
is a Prefect task, or wrap it:
Copy code
@task
  def search_task(task_input, user, password):
      t = Timocom(build_pg_utils_from_block(...), user, password)
      return t.search(task_input)

  search_futures = search_task.map(search_tasks, unmapped(user), unmapped(password))
  for fut in as_completed(search_futures):
      res = fut.result(raise_on_failure=False)
      ...
If it’s still not working, can you share: - How you imported 
as_completed
- The type of 
search_futures[0]
- Whether 
timocom_instance.search
is a Prefect 
@task
- What “doesn’t work” means (blocks until all complete? throws an error? yields nothing?) and any traceback References - Run work concurrently - prefect.futures
10 Views
Open in Slack
PreviousNext
Powered by