https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I have a long running flow which exec...
# ask-marvin
b
@Marvin I have a long running flow which executes hundreds/thousands of tasks in a loop using .submit() (concurrency limited by ThreadPoolTaskRunner). The memory consumption slowly increases continuously over time. Give me an example of a flow/task structure where the flow executes tasks concurrently, while minimising memory creep over time.
m
❌ Error: unhandled errors in a TaskGroup (1 sub-exception)
Sorry, I encountered an error while processing your request. Please try again.
b
@Marvin I have a long running flow which executes hundreds/thousands of tasks in a loop using .submit() (concurrency limited by ThreadPoolTaskRunner). The memory consumption slowly increases continuously over time. Give me an example of a flow/task structure where the flow executes tasks concurrently, while minimising memory creep over time.
m
✅ thought for 115.1 seconds
A good pattern is: keep only a bounded number of futures in memory at once, consume results as tasks finish, and avoid building one giant list of futures/results for the whole run. Here’s a solid Prefect 3 example using 
ThreadPoolTaskRunner
that does that.
Copy code
from prefect import flow, task
from prefect.futures import as_completed
from prefect.task_runners import ThreadPoolTaskRunner


@task
def process_item(item: int) -> int:
    # Simulate work
    return item * 2


@task
def write_batch(results: list[int]) -> None:
    # Replace with DB/file/object-store write
    # Important idea: flush results out of memory regularly
    print(f"writing {len(results)} results")


@flow(task_runner=ThreadPoolTaskRunner(max_workers=20))
def process_items(
    items: list[int],
    submit_batch_size: int = 200,
    flush_batch_size: int = 50,
) -> None:
    for i in range(0, len(items), submit_batch_size):
        current_batch = items[i : i + submit_batch_size]

        # Only submit a bounded batch of tasks
        futures = [process_item.submit(item) for item in current_batch]

        # Consume results as they complete instead of waiting for all
        buffer = []
        for future in as_completed(futures):
            result = future.result()
            buffer.append(result)

            # Periodically flush downstream and clear memory
            if len(buffer) >= flush_batch_size:
                write_batch(buffer)
                buffer.clear()

        # Flush any leftovers from this submit batch
        if buffer:
            write_batch(buffer)

        # Drop references so Python can reclaim memory
        del futures
        del buffer
Why this structure helps: * 
submit_batch_size
limits the total number of in-memory futures. * 
as_completed(futures)
lets you process results incrementally instead of storing all results until the end. * 
write_batch(...)
pushes data out of memory regularly. * 
del futures
and 
buffer.clear()
remove references quickly, which helps GC. A common anti-pattern is this:
Copy code
@flow(task_runner=ThreadPoolTaskRunner(max_workers=20))
def bad_pattern(items: list[int]):
    futures = [process_item.submit(item) for item in items]
    results = [f.result() for f in futures]
    return results
That tends to creep upward because: * you retain all futures until the end * then you retain all results too * for thousands of tasks, that can become a lot of memory pressure A few extra suggestions that usually help: Keep task return values small If each task returns a large dataframe/object/blob, memory creep will be worse. Prefer returning small metadata and writing large outputs to external storage inside the task. Use batching at the flow level Even if your thread pool is only 20 workers, submitting 100k futures in one loop still means the flow is holding references to a huge number of futures. Avoid accumulating logs/data structures in the flow Sometimes the growth is not Prefect itself, but your own growing lists/dicts/caches. If you want an even more isolated pattern for very long runs, use a subflow per batch so each batch gets its own runner lifecycle: ``` from prefect import flow, task from prefect.futures import as_completed from prefect.task_runners import ThreadPoolTaskRunner @task def process_item(item: int) -> int: return item * 2 @task def write_batch(results: list[int]) -> None: print(f"writing {len(results)} results") @flow(task_runner=ThreadPoolTaskRunner(max_workers=20)) def process_batch(items: list[int], flush_batch_size: int = 50) -> None: futures = [process_item.submit(item) for item in items] buffer = [] for future in as_completed(futures): buffer.append(future.result()) if len(buffer) >= flush_batch_size: write_batch(buffer) buffer.clear() if buffer: write_batch(buffer) @flow
def process_items_in_subflows(items: list[int], batch_size: int = 200) -> None: for i in range(0, len(items), batch_size): process_batch(items[i : i + batch_size]) ``` This can help if memory growth comes from long-lived flow/task-runner state over many thousands of task submissions. Practical rule of thumb: * Best for memory: submit a bounded batch → consume with 
as_completed
→ flush externally → clear references * Worst for memory: submit everything → wait at the end → collect all results in one list Relevant docs: * Task runners * Run work concurrently * Flows If you want, I can also give you: 1. a version that writes large task outputs to result storage, or 2. a version instrumented with `psutil`/`tracemalloc` so you can watch memory per batch.
b
@Marvin some of my tasks take and return dataframes. I noticed that if lots of tasks fail, memory consumption increases faster compared to if all the tasks pass. I have also put cache_result=False in all @task decorators, will this make a difference?
m
❌ Error: unhandled errors in a TaskGroup (1 sub-exception)
Sorry, I encountered an error while processing your request. Please try again.
b
@Marvin some of my tasks take and return dataframes. I noticed that if lots of tasks fail, memory consumption increases faster compared to if all the tasks pass. I have also put cache_result=False in all @task decorators, will this make a difference? Currently my flow looks something like this:
Copy code
future_to_id: dict = {}
for id in files: # list of thousands of file ID's
    future = process_file.with_options(
        task_run_name=f"file-id-{id}"
    ).submit(
        file_id=id,
    )
    future_to_id[future] = id

# collect results as they complete
results: list[dict] = []
for f in as_completed(list(future_to_id)):
    activity_id = future_to_id.pop(f)
    try:
        f.result()
        result = {
            "activity_id": activity_id,
            "status": "Success",
            "error": None,
        }
    except Exception as e:
        result = {
            "activity_id": activity_id,
            "status": "Failed",
            "error": str(e),
        } 
    results.append(result)
m
❌ Error: unhandled errors in a TaskGroup (1 sub-exception)
Sorry, I encountered an error while processing your request. Please try again.
b
@Marvin some of my tasks take and return dataframes. I noticed that if lots of tasks fail, memory consumption increases faster compared to if all the tasks pass. I have also put cache_result=False in all @task decorators, will this make a difference? Currently my flow looks something like this:
Copy code
future_to_id: dict = {}
for id in files: # list of thousands of file ID's
    future = process_file.with_options(
        task_run_name=f"file-id-{id}"
    ).submit(
        file_id=id,
    )
    future_to_id[future] = id

# collect results as they complete
results: list[dict] = []
for f in as_completed(list(future_to_id)):
    activity_id = future_to_id.pop(f)
    try:
        f.result()
        result = {
            "activity_id": activity_id,
            "status": "Success",
            "error": None,
        }
    except Exception as e:
        result = {
            "activity_id": activity_id,
            "status": "Failed",
            "error": str(e),
        } 
    results.append(result)
m
✅ thought for 120.7 seconds
Yes — what you’re seeing is plausible, and your current structure can definitely amplify it. Short version: * failures often retain more memory than successes * 
cache_result=False
is probably not the setting you want in Prefect 3.x * the biggest issue in your example is that you submit all thousands of futures up front and keep references to them until they finish * if failures involve DataFrames, the exception traceback can retain references to large local objects for longer A few important points. 1. Failures can retain traceback frames When a task fails, the exception object can keep its traceback, and traceback frames can reference local variables inside the task. If a local variable is a large dataframe, that can make failed tasks “heavier” in memory than successful ones. So this pattern:
Copy code
@task
def process_file(file_id):
    df = load_big_dataframe(file_id)
    ...
    raise ValueError("bad file")
can use noticeably more memory on failure than on success, because the exception/traceback may still reference 
df
. A very practical mitigation is to avoid raising exceptions while large objects are still in scope. For example:
Copy code
@task
def process_file(file_id):
    df = load_big_dataframe(file_id)
    try:
        result = transform_and_validate(df)
    except Exception as exc:
        # explicitly drop big locals before re-raising
        del df
        raise exc
    del df
    return result
Even better, structure code so failure happens after you’ve reduced the dataframe to something small. 2. 
cache_result=False
is not the main knob here In Prefect 3, the setting that matters for in-memory retention is 
cache_result_in_memory=False
. 
cache_result=False
is not the same thing as “don’t keep results in RAM”. Also, caching and result persistence are related but different concepts. The task options worth considering are:
Copy code
@task(
    cache_result_in_memory=False,
    persist_result=False,
)
def process_file(...):
    ...
What these do: * `cache_result_in_memory=False`: helps avoid Prefect keeping successful results around in its in-memory result cache * `persist_result=False`: avoids persisting the task result to result storage But neither of these fully solves memory held by failed exceptions / traceback state while futures still exist. So: yes, 
cache_result_in_memory=False
can help for successful dataframe-returning tasks, but your faster growth during failures is likely more about exception retention than result caching. 3. Your current flow retains too much at once This is the main problem in your snippet:
Copy code
future_to_id: dict = {}
for id in files:
    future = process_file.with_options(
        task_run_name=f"file-id-{id}"
    ).submit(file_id=id)
    future_to_id[future] = id
Even though your thread pool only runs 
max_workers
concurrently, you still create and retain thousands of futures. Each completed future can retain final state, and failed ones may retain exception data/tracebacks. Then later:
Copy code
results: list[dict] = []
for f in as_completed(list(future_to_id)):
    ...
    results.append(result)
This means: * you hold all futures in 
future_to_id
* you make another list copy with 
list(future_to_id)
* you accumulate all summary records in 
results
The 
results
list is probably small enough not to matter much, but the all-at-once future submission definitely does. 4. Best fix: bounded submission window Instead of submitting all files immediately, keep only a limited number of in-flight futures at a time. Here’s a better structure: ``` from prefect import flow, task from prefect.task_runners import ThreadPoolTaskRunner from prefect.futures import wait @task( cache_result_in_memory=False, persist_result=False, ) def process_file(file_id: str) -> None: df = None try: df = load_dataframe(file_id) transformed = transform_dataframe(df) write_output(file_id, transformed)
finally: # drop references to large objects ASAP del df @task def write_results_batch(batch: list[dict]) -> None: # write to DB/file/object store print(f"writing {len(batch)} results") @flow(task_runner=ThreadPoolTaskRunner(max_workers=16)) def run_files(files: list[str], submit_batch_size: int = 100, result_batch_size: int = 50): results_buffer: list[dict] = [] for i in range(0, len(files), submit_batch_size): batch = files[i:i + submit_batch_size] future_to_id = {} for file_id in batch: future = process_file.with_options( task_run_name=f"file-id-{file_id}" ).submit(file_id=file_id) future_to_id[future] = file_id for f in list(future_to_id): file_id = future_to_id.pop(f) try: f.result() results_buffer.append( { "activity_id": file_id, "status": "Success", "error": None, } ) except Exception as e: results_buffer.append( { "activity_id": file_id, "status": "Failed", "error": str(e), } ) # drop reference to future immediately del f if len(results_buffer) >= result_batch_size: write_results_batch(results_buffer) results_buffer.clear() # batch done: all futures from this batch can be GC'd del future_to_id if results_buffer: write_results_batch(results_buffer)
Copy code
This helps because:

* only `submit_batch_size` futures exist at once
* failed futures are not retained for the whole run
* result summaries are flushed periodically
* large dataframe objects should die sooner if the task code drops references

If you want true completion-order processing, use `as_completed`, but still only per bounded batch:
from prefect import flow, task from prefect.futures import as_completed from prefect.task_runners import ThreadPoolTaskRunner @task( cache_result_in_memory=False, persist_result=False, ) def process_file(file_id: str) -> None: df = None transformed = None try: df = load_dataframe(file_id) transformed = transform_dataframe(df) write_output(file_id, transformed) finally: del transformed del df @task def write_results_batch(batch: list[dict]) -> None: print(f"writing {len(batch)} results") @flow(task_runner=ThreadPoolTaskRunner(max_workers=16)) def run_files(files: list[str], submit_batch_size: int = 100, result_batch_size: int = 50): results_buffer: list[dict] = [] for i in range(0, len(files), submit_batch_size): batch = files[i:i + submit_batch_size] future_to_id = {} for file_id in batch: future = process_file.with_options( task_run_name=f"file-id-{file_id}" ).submit(file_id=file_id) future_to_id[future] = file_id for f in as_completed(list(future_to_id)): file_id = future_to_id.pop(f) try: f.result() results_buffer.append( { "activity_id": file_id, "status": "Success", "error": None, } ) except Exception as e: results_buffer.append( { "activity_id": file_id, "status": "Failed", "error": str(e), } ) finally: del f if len(results_buffer) >= result_batch_size: write_results_batch(results_buffer)
results_buffer.clear() del future_to_id if results_buffer: write_results_batch(results_buffer)
Copy code
*5. If tasks return dataframes, consider not returning them at all*

If possible, this is the biggest improvement:

* task loads dataframe
* task transforms dataframe
* task writes dataframe/output somewhere external
* task returns only small metadata

For example:
@task(cache_result_in_memory=False, persist_result=False) def process_file(file_id: str) -> dict: df = load_dataframe(file_id) out_path = write_transformed_dataframe(transform_dataframe(df), file_id) del df return {"file_id": file_id, "out_path": out_path}
Copy code
Better than:
@task def process_file(file_id: str) -> pd.DataFrame: df = load_dataframe(file_id) return transform_dataframe(df)
Copy code
Returning dataframes makes both success and failure paths heavier.

*6. A subtle improvement for failure-heavy tasks*

If you only need failure info in the parent flow, consider catching exceptions *inside the task* and returning a small status object instead of letting the task fail with a traceback.

Example:
@task(cache_result_in_memory=False, persist_result=False) def process_file(file_id: str) -> dict: df = None transformed = None try: df = load_dataframe(file_id) transformed = transform_dataframe(df) write_output(file_id, transformed) return { "activity_id": file_id, "status": "Success", "error": None, } except Exception as e: return { "activity_id": file_id, "status": "Failed", "error": str(e), } finally: del transformed del df
Copy code
Then the flow becomes:
@flow(task_runner=ThreadPoolTaskRunner(max_workers=16)) def run_files(files: list[str], submit_batch_size: int = 100): for i in range(0, len(files), submit_batch_size): batch = files[i:i + submit_batch_size] futures = [process_file.submit(file_id=file_id) for file_id in batch] batch_results = [] for f in as_completed(futures): batch_results.append(f.result()) write_results_batch(batch_results)
Copy code
Why this can help:

* successful and failed tasks both return small dicts
* you avoid retaining Python exception tracebacks in failed future states
* memory is often much more stable

Tradeoff:

* tasks won’t be marked failed in Prefect UI unless you explicitly choose to raise
* you’d be treating failure as data, not orchestration failure

That may or may not fit your use case.

*My recommendation for your case*

I would try these in order:

1. Change task decorators to use:
@task(cache_result_in_memory=False, persist_result=False) ``` 2. Stop submitting all thousands at once; use bounded batches. 3. Avoid returning dataframes from tasks if possible. 4. In task code, explicitly drop dataframe references in 
finally
. 5. If many failures are expected and you mostly need reporting, catch exceptions inside the task and return a small status dict instead of failing the task. Useful docs: * Results and result persistence * Run work concurrently * Task runners If you want, I can rewrite your exact snippet into: * a bounded in-flight futures version, or * a failure-as-data version optimized for dataframe-heavy tasks.
b
@Marvin I tried using batching, but it didn't make a difference to continuously rising memory usage. Rather than using ThreadPoolTaskRunner, could I use ProcessPoolTaskRunner instead? Would the memory consumed by the tasks be released after they complete?
m
❌ Error: unhandled errors in a TaskGroup (1 sub-exception)
Sorry, I encountered an error while processing your request. Please try again.
b
@Marvin I tried using batching, but it didn't make a difference to continuously rising memory usage. Rather than using ThreadPoolTaskRunner, could I use ProcessPoolTaskRunner instead? Would the memory consumed by the tasks be released after they complete?
m
✅ thought for 103.2 seconds
Yes, you can try 
ProcessPoolTaskRunner
, but the answer to “will memory be released after each task completes?” is: not necessarily. With a process pool, worker processes are typically reused across many tasks for the lifetime of the runner. So memory used inside a worker may stay associated with that worker process until: * the worker process exits, or * Python in that process actually returns memory to the OS, which is not guaranteed So 
ProcessPoolTaskRunner
can help in some cases, but it is not the same as “fresh process per task”. What it can help with ---------------------- If your memory issue is caused by: * large temporary objects living inside task execution * pandas / numpy allocations inside workers * GIL-related thread contention for CPU-heavy dataframe transforms then process-based execution can sometimes behave better than threads, because each worker has its own memory space. But there are important tradeoffs. Big caveat for dataframe-heavy tasks ------------------------------------ If you use 
ProcessPoolTaskRunner
, task inputs and outputs need to be serialized between processes. That means if your tasks: * take a dataframe as input, or * return a dataframe as output you may actually make memory behavior worse, because large objects are copied/serialized across process boundaries. So this is usually a bad fit:
Copy code
@task
def process_df(df):
    return transform(df)
with a process pool, because you are shipping whole dataframes between processes. A much better fit is:
Copy code
@task
def process_file(file_id):
    df = load_dataframe(file_id)
    transformed = transform(df)
    write_output(file_id, transformed)
    return {"file_id": file_id, "status": "Success"}
That way the worker: * loads the dataframe locally * processes it locally * writes it externally * returns only a small dict That’s the pattern where 
ProcessPoolTaskRunner
is most likely to help. Will memory be freed after tasks complete? ------------------------------------------ Usually: * memory for Python objects that become unreachable can be reclaimed inside the worker process * but the worker process itself remains alive * and many allocators do not eagerly return freed memory to the OS So you may see: * task-local memory usage goes down somewhat * but RSS of the worker process stays high * total process-pool memory plateaus instead of dropping fully That is normal for process pools. So if your question is: “Will each task fully release its memory back to the OS when done?” the realistic answer is: generally no, not reliably. When process pools help most ---------------------------- 
ProcessPoolTaskRunner
is most promising when: * tasks are CPU-bound * tasks load data from files/object storage themselves * tasks return only small summaries/status objects * task failures are caused by computation/validation, not by carrying giant dataframes back out * you want isolation from the main flow process When process pools help least ----------------------------- They are less likely to help if: * you pass large dataframes into tasks * you return large dataframes from tasks * failures hold big exceptions/tracebacks referencing local dataframe objects * the worker processes stay alive for a very long time and keep high-water-mark memory Recommended pattern if you try it --------------------------------- If you want to experiment, I’d structure it like this: ``` from prefect import flow, task from prefect.task_runners import ProcessPoolTaskRunner from prefect.futures import as_completed @task( cache_result_in_memory=False, persist_result=False, ) def process_file(file_id: str) -> dict: df = None transformed = None try: df = load_dataframe(file_id) transformed = transform_dataframe(df) write_output(file_id, transformed) return { "activity_id": file_id, "status": "Success",
"error": None, } except Exception as e: return { "activity_id": file_id, "status": "Failed", "error": str(e), } finally: del transformed del df @task def flush_results(results: list[dict]) -> None: print(f"flushing {len(results)} results") @flow(task_runner=ProcessPoolTaskRunner(max_workers=4)) def run_files(files: list[str], batch_size: int = 50, flush_size: int = 25): buffer = [] for i in range(0, len(files), batch_size): batch = files[i:i + batch_size] futures = [process_file.submit(file_id) for file_id in batch] for f in as_completed(futures): buffer.append(f.result()) if len(buffer) >= flush_size: flush_results(buffer) buffer.clear() del futures if buffer: flush_results(buffer) ``` Why this is a reasonable test: * workers load the dataframe themselves * tasks return only small dicts * failures are converted to small result payloads instead of large traceback-bearing failed states * concurrency is bounded Important practical note ------------------------ If your primary goal is memory release after each unit of work, a process pool may still not be enough. What usually gives the strongest memory reset is: shorter-lived subprocess lifecycles In Prefect terms, that often means: * splitting work into smaller flow runs / subflows * letting each run use a fresh task runner / fresh worker processes * ending that run so the processes exit That gives a harder reset than keeping one long-lived process pool alive for hours. So if batching within one flow did not help, the next thing to try is not just “switch threads to processes”, but: split work into multiple shorter flow runs or subflows so the pool is torn down periodically. My recommendation ----------------- If your tasks are dataframe-heavy, I’d rank the options like this: 1. Best: task loads dataframe itself, writes output externally, returns small metadata 2. Then: use 
ProcessPoolTaskRunner
if tasks are CPU-heavy 3. Even better for hard memory resets: break into shorter subflows / separate flow runs so the process pool is recreated periodically 4. Avoid: passing or returning large dataframes between flow and task boundaries So yes, it is worth trying 
ProcessPoolTaskRunner
, but only if you also change the task contract so that dataframes stay inside the task process. Docs: * Task runners * Run work concurrently * Results and result persistence If you want, I can rewrite your current 
process_file
pattern into: * a 
ProcessPoolTaskRunner
version optimized for dataframe workloads, or * a subflow-per-batch pattern that forces more reliable memory release.
n
cache_result_in_memory=False
can help when successful tasks return large objects, but it will not fully address memory growth from failed tasks. for failures, one important factor is traceback retention. when a task raises, the exception keeps its traceback, and traceback frames can keep local variables alive. If a dataframe is still in local scope when the task raises, the failed task state/future can retain that dataframe longer than expected looking at what you have, i'd try 3 things 1. submit work in bounded groups instead of creating thousands of futures at once. 2. avoid returning dataframes from tasks; write large outputs inside the task and return small metadata. 3. if a task may fail after creating a dataframe, either handle the failure inside the task and return a small failure record, or clear large locals before re-raising. example failure-as-data pattern:
Copy code
@task(cache_result_in_memory=False, persist_result=False)
def process_file(file_id: str) -> dict:
    df = None
    try:
        df = load_dataframe(file_id)
        output_path = write_processed_dataframe(file_id, df)
        return {"activity_id": file_id, "status": "Success", "output": output_path, "error": None}
    except Exception as exc:
        return {"activity_id": file_id, "status": "Failed", "output": None, "error": str(exc)}
    finally:
        del df
if you need the task marked failed in the API, re-raise after releasing large locals:
Copy code
except Exception:
    del df
    raise
then collect results in bounded batches rather than keeping all futures/results around for the whole flow run
🙏 1
b
@Marvin Would using ProcessPoolTaskRunner or DaskTaskRunner help reclaim memory from each file processing task?
m
✅ thought for 26.1 seconds
Potentially, but only under certain conditions. Short answer: * `ProcessPoolTaskRunner`: maybe somewhat, but not reliably per task * `DaskTaskRunner`: maybe, but for the same reason, not automatically per task * neither guarantees “memory is reclaimed after every file task” * the strongest memory reset comes from ending the worker process itself, not just finishing a task Here’s the practical distinction. ProcessPoolTaskRunner ----------------------- 
ProcessPoolTaskRunner
uses a pool of long-lived worker processes. That means: * a task runs in a worker process * when the task finishes, Python objects may be freed in that process * but the worker process stays alive for future tasks * memory often stays reserved by that process allocator, even if objects are gone So it can help if: * the problem is specific to thread-shared memory behavior * your task loads/processes/writes data entirely inside the worker * the task returns only small metadata But it usually does not mean: * “after each file task, all memory goes back to the OS” It is more like: * “memory is isolated per worker, and may stabilize better than threads in some workloads” DaskTaskRunner ---------------- 
DaskTaskRunner
also uses long-lived workers. So the same core limitation applies: * task finishes * worker stays alive * memory may or may not be returned to the OS * workers can accumulate memory fragmentation / high-water marks over time Dask can sometimes behave better for memory management than a simple local process pool because it has: * a separate scheduler/worker model * worker-level memory management policies * the ability to spill data to disk in some configurations * better observability into worker memory But for your specific question — “reclaim memory from each file processing task?” — the answer is still: not reliably per task unless workers themselves are restarted or their memory policy forces cleanup. So if your mental model is: task finishes => worker fully clears memory that is usually not true for either ProcessPool or Dask. When they do help ------------------- They can help if your current issue is mostly due to: * large temporary pandas/numpy allocations inside threads * failures retaining heavy traceback/object graphs in the main process * CPU-bound transforms that are better isolated in separate processes * not wanting one process heap to absorb all task allocations They help most when each task: * reads the file itself * processes the dataframe internally * writes output externally * returns only a small dict/status Example of the right shape:
Copy code
@task
def process_file(file_id: str) -> dict:
    df = load_dataframe(file_id)
    out = transform_dataframe(df)
    write_output(file_id, out)
    return {"file_id": file_id, "status": "ok"}
Example of the wrong shape for process-based runners:
Copy code
@task
def process_file(df):
    return transform_dataframe(df)
because now you are shipping large dataframes between processes/workers. What actually helps memory reclamation most ------------------------------------------- If your goal is real reclamation after each file or each small group of files, the best options are usually: * a fresh process for each unit of work * or restarting workers periodically * or splitting work into smaller flow runs / subflows that end completely In practice: * best reclamation: separate process lifecycle per chunk * medium: process pool / dask worker pool * least: thread pool in one long-lived process So between your two options: * 
ProcessPoolTaskRunner
may help some * 
DaskTaskRunner
may help some, and gives more scaling/memory tooling * but neither is a guaranteed fix for monotonic memory growth My practical recommendation --------------------------- If you want to test runner choice specifically: Try 
ProcessPoolTaskRunner
first if: * you are on one machine * tasks are CPU-heavy / pandas-heavy
* each task can load its own file and return tiny metadata Try 
DaskTaskRunner
 if: * you already use Dask * you want worker-level memory controls / scaling * you want to isolate work across a more robust worker system But if your top priority is memory reset, I would prioritize architecture over runner: 1. keep dataframes inside the task 2. return only tiny results 3. catch expected failures inside the task and return failure metadata instead of raising 4. process files in smaller chunks with shorter-lived worker lifecycles 5. if needed, use subprocess/process-worker boundaries that actually terminate periodically Rule of thumb ------------- If you ask: “Will this reclaim memory after each file task?” Then: * `ThreadPoolTaskRunner`: mostly no * `ProcessPoolTaskRunner`: sometimes partially, not reliably * `DaskTaskRunner`: sometimes partially, with more controls, not reliably * fresh worker/process per chunk: yes, most reliably If you want, I can give you: * a decision table for ThreadPool vs ProcessPool vs Dask for pandas workloads, or * a concrete Prefect pattern that forces worker/process turnover every N files for better memory reclamation.
7 Views
Open in Slack
PreviousNext
Powered by