Hi there, I'm using Hamilton's caching in conjunction with

Parallelizable

. This leads to unexpectedly long execute times, even though the data was in fact restored from the cache. I suspect this is due to the number of nodes that are being checked. Essentially, I have a parallel Node that for a list of directories goes into the directory and processes some data to return a DataFrame. All DataFrames then get collected into one big DataFrame in the end. What are your recommendations? Is there a way for me to give hamilton more information about the data such that it can speed up cache lookups? And is there the opposite of

@cache(behavior="recompute")

to tell hamilton that something is assumed to be up-to-date? Doing

id = dr.cache.last_run_id
data_version = dr.cache.data_versions[id]['all_frames']
result_store.get(data_version)

is an option, but it would still be nice to speed up

dr.execute

when the cache is up-to-date.

CC @Thierry Jean Interesting. Yeah we'd have to profile the code a little to know where things are slowing down. It could be that the SERDE costs (or creating hashes) for each file is the cause -- how many files do you have approximately (order of magnitude) to process?

Hi @Jannik Buhr! Caching in Hamilton has good defaults and many configuration points. Caching is a hard problem and it might need some tuning for your use case. I'll give you a few pointers! ## Assume result is up-to-date > And is there the opposite of

@cache(behavior="recompute")

to tell hamilton that something is assumed to be up-to-date? Hamilton's caching is entirely determined by code and data versions. It should automatically figure out if values are "up-to-date" (manual overrides by users would very likely break things). Can you share the DAG shape that causes problems? Remember that you can always use the simple `dr.execute([...], overrides={...})`In your case, it could look like:

dr = Builder().with_modules().with_cache().build()
# execution 1
final_vars = [...]
dr.execute(final_vars)

# do some manual checks on the cache
id = dr.cache.last_run_id
data_version = dr.cache.data_versions[id]['all_frames']
all_frames = result_store.get(data_version)

dr.execute(final_vars, overrides={"all_frames": all_frames})

(we should create a utility to simplify this pattern) Here are the details for all caching behaviors. This section of the caching tutorial gives example use cases for caching behaviors. ## Parallel execution is throttling my CPU I would first investigate how the issue ties with

Parallelizable

. You can try setting an executor that only executes nodes in sequence (documentation):

from hamilton.driver import Builder
from hamilton.execution import executors

dr= (
  Builder()
  .enable_dynamic_execution(allow_experimental_mode=True)
  .with_modules(...)
  # local executor applies to "nodes outside Parallel/Collect" branches
  .with_local_executor(executors.SynchronousLocalTaskExecutor())
  # remote executor applies to "nodes inside Parallel/Collect" branches
  .with_remote_executor(executors.SynchronousLocalTaskExecutor())
  # or, you can set a max number of 
  # .with_remote_executor(executors.MultiProcessingExecutor(max_tasks=5))
  .with_cache()
  .build()
)

Using this code should remove the potential factor "my CPU is running too many things at once". Python is hard to optimize and if you're using Polars, numpy, scikit-learn, etc. their own optimization might conflict with your Hamilton DAG optimizations, throttling the CPU ## Dataframes are blowing up RAM

pd.DataFrame

are inefficient in terms of memory-space and they allow some operations that are very expensive (doing the operation requires more temporary memory). If you do

Parallel/Collect

your RAM may be getting full of

pd.DataFrame

. The previous code using

SynchronousLocalTaskExecutor()

doesn't solve this. Even if you execute one-by-one, you need to collect them all in to memory. The best solution would be to not use

pandas

. If you do, the smallest refactor would be to store these small dataframes as parquet files, then write a node that does the join operation using

duckdb

from the parquet files and returns the joined df. ## Caching serialization/deserialization is expensive It becomes inefficient to serialize large dataframes using the default

pickle

module (transform a dataframe as bytes); Setting the cache format to parquet would help ## Caching fingerprinting is expensive Caching requires fingerprinting, i.e., uniquely identifying a Python object. This is a challenging task, which can be costly for dataframes. We have default implementations, but you can register your own one. (I believe it's unlikely to be the bottleneck)

Thanks for all the tips! I created a a small toy pipeline to experiment: https://github.com/jmbuhr/hamilton-test, with the result rendered here: https://jmbuhr.de/hamilton-test/. • In this particular example, doing the

Parallelizable

itself adds more overhead (even when the executor is set to be synchronous) than it makes up in time. This may not necessarily be the case in my real case, because more happens to the data within each task, but it's something to keep in mind • But the question if cache checking (in the parallelizable case) could be sped up still applies here. I think this confusion stems from me thinking from left to right i.e.

dir

didn't change and neither did any downsteam code, so it is obvious that there is no need to recompute

one_frame

and

result

. But the computer starts at the requested

result

and because

one_frame

is downstream of a Parallelizable (

dir

) it is expanded into 130 Nodes, which all have to be checked individually, even though as the human in the loop we know that none of them can have changed, because the upstream

dir

didn't change.

Thanks for the repro! Some other thoughts • Caching is implemented a lifecycle hook, so we always need to traverse all nodes. Caching only determines if the node is executed or not •

Parallelizable/Collect

allows you to create

n

branches of the DAG, where

n

is only known at runtime. For caching, we must a. execute

dir

to determine how many branches b. evaluate caching conditions for nodes inside

Parallelizable/Collect

for each branch • The situation you describe assumes that you know

n

before execution The behavior you describe would require caching conditions to be "planned" rather than "node-by-node". It's something that was considered, but making this robust will likely have to wait for Hamilton 2.0 and core changes

thanks! This all makes sense, so I ended up parallelizing within a node using standard python multiprocessing instead of creating all those nodes.