https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> why is ram from tasks not released ev...
# ask-marvin
o
@Marvin why is ram from tasks not released even though gc.collect() and del of objects is applied inside task level code?
m
✅ thought for 95.8 seconds
Short version: 
del
and 
gc.collect()
only remove Python references they can actually reach. In Prefect, task memory often stays high because the process still holds references through task state, futures, exceptions/tracebacks, thread pools, logging queues, or native libraries. Also, even when Python objects are freed, CPython may not return that memory to the OS immediately. A few common reasons in Prefect 3.x: 1. The task result is still being retained - Prefect stores task outcomes in 
State
objects. - If you return a large object, Prefect may still hold it via the task state and future. - If your flow does:
Copy code
fut = my_task.submit()
  result = fut.result()
then both 
fut
and 
result
can keep memory alive until they go out of scope. 2. Exceptions keep locals alive - If a task errors, the exception traceback can retain references to local variables. - This is a classic Python behavior: traceback -> frame -> locals -> large objects. - Prefect also tracks failures in task state, so failed tasks can hold onto more memory than successful ones. 3. Thread-based execution doesn’t isolate memory - With the default thread-based execution, tasks run in the same Python process. - If memory is fragmented or held by allocators, the process RSS may not drop even after objects are freed. - So “memory not released” may mean “Python can reuse it later,” not “the objects still exist.” 4. Native libraries may keep allocations - Pandas, NumPy, PyArrow, Torch, etc. often allocate memory outside normal Python object management. - 
gc.collect()
only affects Python-tracked objects, not necessarily memory held by native allocators. 5. Logging / queued internal objects - Prefect keeps task metadata, states, and logs around long enough to report orchestration status. - That can delay reclamation. What usually helps most Use smaller task outputs - Don’t return huge dataframes/arrays if you can persist them externally and return a path, key, or summary instead. Drop futures/results promptly
Copy code
@flow
def f():
    fut = heavy_task.submit()
    x = fut.result()

    # use x

    del x
    del fut
    import gc
    gc.collect()
Prefer process isolation for memory-heavy work - If a task is truly memory-intensive and you need memory to be actually reclaimable, run it in a separate process/container. - When that process exits, the OS reclaims the memory reliably. - In practice this often means: - using infrastructure/workers that execute flow runs in separate processes/containers - or using process-based parallelism instead of threads for the heavy step Persist large intermediates outside memory - Write to object storage / parquet / database, and pass references between tasks. Be careful with failed tasks - Large objects created before an exception may stay alive through traceback retention. - Catching exceptions earlier and reducing what’s in scope can help. A useful mental model: - 
del obj
= remove one reference name - 
gc.collect()
= collect unreachable Python objects - Neither guarantees: - Prefect isn’t holding another reference - native memory is released - OS-visible RSS drops immediately If you want, I can help you diagnose this in a *Prefect-specific pattern*—for example: - task returns large dataframe - many 
.submit()
calls in a loop - mapped tasks - failures causing memory growth - Dask/Ray/task runner setup If you share a small flow/task example, I can point to the exact retention path.
o
@Marvin how to run only one task in a seperate process but otherwise keep threaded exection for the flow and all other tasks?
m
✅ thought for 153.5 seconds
Yes — but not directly on a single 
@task
via a task-runner setting. In Prefect 3.x, task runners are configured at the 
@flow
level, not per task. So if your main flow uses 
ThreadPoolTaskRunner
, then tasks submitted with 
.submit()
in that flow use threads. Best supported pattern: put that one task inside a subflow that uses 
ProcessPoolTaskRunner
, while the parent flow keeps 
ThreadPoolTaskRunner
. Example:
Copy code
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner, ProcessPoolTaskRunner

@task
def normal_task(x):
    return x * 2

@task
def memory_heavy_task(x):
    # runs in separate process when called from process subflow
    big = [x] * 10_000_000
    return sum(big)

@flow(task_runner=ProcessPoolTaskRunner(max_workers=1))
def isolated_process_subflow(x):
    fut = memory_heavy_task.submit(x)
    return fut.result()

@flow(task_runner=ThreadPoolTaskRunner(max_workers=8))
def main_flow():
    a = normal_task.submit(1)
    b = normal_task.submit(2)

    isolated = isolated_process_subflow(5)

    return {
        "a": a.result(),
        "b": b.result(),
        "isolated": isolated,
    }
What this gives you: - 
main_flow
tasks run with threads - 
memory_heavy_task
runs in a separate process - that process-based runner is scoped to the subflow Important caveat: - the subflow call is synchronous from the parent’s perspective unless you structure around it - so this is good for isolation, but not a per-task runner override inside the same submitted task set Alternative options 1. Launch a subprocess manually inside one task If you need true one-off OS-level isolation for exactly one operation, you can do:
Copy code
from prefect import task
import subprocess
import sys

@task
def run_heavy_thing():
    result = subprocess.run(
        [sys.executable, "heavy_script.py"],
        capture_output=True,
        text=True,
        check=True,
    )
    return result.stdout
This is often the strongest memory-isolation option because when the subprocess exits, its RAM is fully reclaimed by the OS. 2. Use a separate deployed flow / worker If the isolated step is big enough, turn it into its own flow/deployment and call it separately. That gives stronger isolation boundaries, but it’s more operational overhead. What you cannot do There is no supported pattern like:
Copy code
@task(task_runner=ProcessPoolTaskRunner())
def one_task():
    ...
That parameter does not exist on 
@task
. Also relevant: the 
@task
signature does not include a 
task_runner
argument, which confirms this. If you want, I can show you: - a version where the threaded parent flow continues doing work while the isolated subflow runs - or a subprocess wrapper task pattern specifically for memory-hungry pandas/numpy code.
o
@Marvin is there a known pattern on how to release and remove artefacts withouth usind ProcessPoolTaskRunner?
m
✅ thought for 122.0 seconds
Yes — there are known patterns, but the important distinction is: If by “artifacts” you mean Python objects / task results in memory, then the pattern is about avoiding retention, not forcing the thread-based worker process to give RAM back to the OS. If by “artifacts” you mean Prefect Artifacts (markdown/table/progress objects in the UI), those are separate and can be deleted, but they are usually not the cause of RAM growth. For your memory case, the main patterns without 
ProcessPoolTaskRunner
are: --- 1. Do not keep large task results in Prefect memory Use:
Copy code
@task(
    persist_result=False,
    cache_result_in_memory=False,
)
def heavy_task(...):
    ...
Or, if you do need persistence:
Copy code
@task(
    persist_result=True,
    cache_result_in_memory=False,
)
def heavy_task(...):
    ...
Why this helps: - 
persist_result=False
means Prefect does not persist the return value as a result - 
cache_result_in_memory=False
means Prefect avoids retaining the result in its in-memory result cache This is probably the most important Prefect-level knob. --- 2. Return a small handle, not the large object Instead of returning a dataframe / array / model, write it somewhere and return a path/key:
Copy code
from prefect import task

@task(cache_result_in_memory=False)
def build_big_df():
    df = ...
    path = "/tmp/output.parquet"
    df.to_parquet(path)

    del df
    return {"path": path}
Then downstream tasks load from that path only if needed. This is the most common pattern for thread-based flows. --- 3. Consume futures/results quickly, then delete them If you keep a list of many futures or results, you keep memory alive. Bad:
Copy code
futures = [heavy_task.submit(x) for x in items]
results = [f.result() for f in futures]
Better:
Copy code
for x in items:
    fut = heavy_task.submit(x)
    result = fut.result()

    # use result

    del result
    del fut
Or process in small batches:
Copy code
batch_size = 20
for i in range(0, len(items), batch_size):
    batch = items[i:i+batch_size]
    futures = [heavy_task.submit(x) for x in batch]

    for fut in futures:
        result = fut.result()
        # use result
        del result

    del futures
--- 4. Delete large locals before raising exceptions This is a big one for memory leaks that look mysterious. If a task fails, Python tracebacks can retain all locals in the frame. Bad:
Copy code
@task
def heavy_task():
    big = make_huge_object()
    raise ValueError("boom")
Better:
Copy code
@task
def heavy_task():
    big = make_huge_object()
    try:
        do_work(big)
    except Exception:
        del big
        raise
Also useful:
Copy code
@task
def heavy_task():
    big = None
    try:
        big = make_huge_object()
        return do_work(big)
    finally:
        if big is not None:
            del big
That won’t solve all cases, but it helps a lot when failures are involved. --- 5. Keep object scope as small as possible Instead of one huge task that loads, transforms, aggregates, and returns a giant object, split it so the large object dies earlier. Example:
Copy code
@task(cache_result_in_memory=False)
def extract_to_file():
    ...
    return {"path": "/tmp/raw.parquet"}

@task(cache_result_in_memory=False)
def transform_file(meta):
    ...
    return {"path": "/tmp/transformed.parquet"}

@task
def summarize(meta):
    ...
    return {"row_count": 123}
This reduces the lifetime of large in-memory objects. --- 6. Prefer external storage for intermediates Common pattern: - parquet on disk - object storage - temp tables - warehouse staging tables Then Prefect coordinates small references, not huge Python objects. --- 7. Be aware that 
gc.collect()
does not mean RSS goes down Even if all references are gone: - CPython may keep arenas allocated - NumPy/pandas/pyarrow/native allocators may keep memory - thread-based execution keeps everything in one long-lived process
So a successful pattern is often: - make memory reusable within the process - not necessarily expect OS-visible RAM to drop immediately That means if later tasks do more heavy work, they may reuse that memory. --- 8. Prefect Artifacts are different If you literally mean Prefect Artifacts created with things like markdown/table/progress artifacts: - those are API/UI metadata - they are not the main driver of in-process RAM retention They can be deleted via the API or CLI, but that’s usually unrelated to Python RAM. If needed, artifact deletion is available through the client or CLI. But for your issue, the real culprit is much more likely: - task result retention - future/state retention - traceback retention - large native allocations --- Recommended thread-only pattern If you want to stay threaded, the most practical pattern is:
Copy code
from prefect import flow, task
import gc

@task(persist_result=False, cache_result_in_memory=False)
def heavy_task(x):
    big = [x] * 10_000_000
    total = sum(big)
    del big
    gc.collect()
    return total

@flow
def main(items):
    outputs = []

    for x in items:
        fut = heavy_task.submit(x)
        result = fut.result()

        outputs.append(result)

        del result
        del fut
        gc.collect()

    return outputs
And if the returned value is also large, replace it with: - a file path - object storage key - DB table name - compact metadata --- Most effective knobs to try first 1. 
cache_result_in_memory=False
2. 
persist_result=False
for intermediates 3. return references, not large objects 4. avoid accumulating many futures/results 5. delete large locals before exceptions escape If you want, I can give you a concrete Prefect 3 memory-safe task pattern for: - pandas dataframe tasks - numpy arrays - LLM/model inference outputs - batch loop flows
3 Views
Open in Slack
PreviousNext
Powered by