Adds read_after_write flag Rather than using persist with pyspark, you could just tell Spark to read from it.
diff --git a/hamilton/experimental/h_cache.py b/hamilton/experimental/h_cache.py index 18b24ba..ba0a74c 100644 --- a/hamilton/experimental/h_cache.py +++ b/hamilton/experimental/h_cache.py
@@ -298,6 +298,7 @@ writers: Optional[Dict[str, Callable[[Any, str, str], None]]] = None, readers: Optional[Dict[str, Callable[[Any, str], Any]]] = None, read_kwargs: Optional[Dict[str, Any]] = None, + read_after_write: bool = False, **kwargs, ): """Constructs the adapter. @@ -318,7 +319,7 @@ self.readers = readers or {} self.read_kwargs = read_kwargs or {} - + self.read_after_write = read_after_write self._init_default_readers_writers() def _init_default_readers_writers(self): @@ -385,6 +386,10 @@ cache_format, ) self._write_cache(cache_format, result, filepath, node.name) + if self.read_after_write: + # this could be useful for delayed execution type things as a means to reset + # that they have set internally + result = self._read_cache(cache_format, result, filepath) self.computed_nodes.add(node.name) return result empty_expected_type = self._get_empty_expected_type(node.type)