pennmem / cml_pipelines Goto Github PK

A lot of times when building pipelines, we might want to do something like this:

results = [func(x) for x in iterable]
return results

To make this work easier with delayed tasks, the Pipeline class should gain a sink method that collects the results like this for us so that we can use that to return a single dask.Delayed instance in the build method like so:

def build(self):
    results = [self.delayed_func(x) for x in self.values]
    return self.sink(results)

We should do one of two things upon successful completion of a pipeline:

1. Automatically clear the cache
2. Add a decorator to do this which wraps a function that runs a pipeline.

The joblib caching mechanism leaves much to be desired. We should just get rid of it and use the dask.distributed decorator directly.

Futures are great, but in most cases when we're building pipelines, we need to wait on the results so blocking should be the default.

A better way to implement the FilePaths class is to subclass dict and override setitem similar to tornado's ObjectDict implementation: https://github.com/tornadoweb/tornado/blob/81eb9f75238c1e3442f1687d0ca1030c5ba3f177/tornado/util.py#L97-L109

Right now if you want to clear the cache manually, you have to do from cml_pipelines.wrapper import memory and then memory.clear() which is kind of awkward. We should come up with a friendlier way to do this (and set cache directory, etc.).