Dynamic Execution

[elijahbenizzy]

With the next release of Hamilton, you’ll be able to dynamically rerun portions of the DAG for a set of inputs, in parallel! What does this look like?

def file_to_process() -> Parallelizable[str]:
    """Returns a list of files to process"""
    return [item for item in _list_files()]

def file_loaded(file_to_process: str) -> str:
    return _load (file_to_process)

def word_count(file_loaded: str) -> str:
    return len(file_loaded.split(" "))

def all_word_count(word_count: Collect[int]) -> int:
    return sum(word_count)

So, a simple map-reduce. Then you’ll be able to specify executors in the driver, which will do two things:

• group nodes into “tasks” (to avoid serialization boundaries) — in this file_loaded and word_count form the same task
• execute them, in parallel, on the executor of your choice (fully pluggable)!

Use-cases are:

• call out to a set of models in parallel over different input
• process a bunch (previously unknown #) of files in parallel
• run a portion of the DAG over different parameters that aren’t known in advance (hyperparameter optimization, sliding window computation, etc…)
• …

Note that, in the future, we will also be adding the capability to output a generator and run over the subsequent subdags in a serial order (E.G. for mini-batching).

See PR here.

The asks are two-fold:

• *Please help figure out what to call these!
• Please reach out if you have a use-case/want to test! Happy to give you RC access, and help you through any rough spots

[jdonaldson]

Many of the pain points here surround supporting aborted/problematic runs. Are you looking into providing SLA for this, or some kind of recovery mechanism, or is it just raw functionality for now?

[jdonaldson]

e.g. what happens if the third file in my (previously unknown #) of files is faulty? Does this mechanism support halting the rest of the map operation?

[elijahbenizzy]

Thanks for the input! So yes, that's next in the plan, but currently we're providing the raw functionality. This is still in beta (there's an experimental flag), so we have some wiggle room.

Its current approach will halt the rest of the map operation (waiting for in-flight tasks to finish, although that should likely be changed) but ideally this would be (a) customizable, and (b) cacheable.

Some more use-cases --

• hitting a flaky API a bunch -- workflow fails but we want to keep our current results so we can continue until its done
• Allow for optional results/defaults if one of the spawned tasks is allowed to fail

Would love thoughts on APIs around here. I also think the user can build these in with a little cleverness -- E.G. if you save the result of each file that's been processed as part of your subdag that does the processing, you can actually just bypass them and load them up before running the Collect node.

[jdonaldson]

From your twitter post:

E.g. is there a better way to annotate/express that the functions/nodes between file_to_process() and all_word_count() are parameterized on the output of files_to_process? I.e. would you understand that's what Parallelizable and Collect signify here? or not? 2/3

I'd expect to be able to write the function from your example as follows:

def all_word_count(word_count: Collect[list[int]]) -> int:
    ...

To gather the examples, I'd expect to wrap invocations to word_count in a DAG invocation to that branch of the graph. That invocation itself is wrapped in a parrallelizable invocation across files I was interested in. I'd pass the result of that into all_word_count.

Trying to mix dynamism into a DAG means you lose some of the ability to optimize the invocations. But in some cases it's ok. If you add in SLA or some other observability capability it might be nice. However, I imagine it's also going to be a source of bugs for you that you'd rather not deal with.

[elijahbenizzy]

Interesting -- would DAG[...] basically specify a DAG to run?

If so, you might like @subdag -- its effectively that, except done through static parameters. https://hamilton.dagworks.io/en/latest/reference/decorators/subdag/

And yeah, its a question of how dynamic/chaotic we want this to be...

[jdonaldson]

Interesting -- would DAG[...] basically specify a DAG to run?

No, it's basically the type signature for the entire DAG. It's not a replacement for Collect[Collect[int]] though, it's a higher level map of the DAG. I'm realizing it's too high level to talk about right now, but FWIW here's a library that applies the same concept to web routing instead of DAGs: https://github.com/jdonaldson/golgi. FWIW, that library was highly experimental and didn't see much use.

The @subdag concept is a really good first class feature IMHO, it fits really well with the notion of transitive closures. In your visualization/ui, it gives a nice boundary to expand/collapse depending on if you're interested in it or not. I think that's a big win.

Back to Dynamism, I guess I just offer warnings about it because it's so easy for your project to get "tarred and feathered" with any poor performance/functionality that happens as a result. I'd really hate to see that happen, it's totally unfair. It would be great to see the scenarios where people are wanting Dynamism, and offering first class fixtures for that.

E.g. for universal file based IO scenarios, I really like fsspec : https://filesystem-spec.readthedocs.io/en/latest/

[elijahbenizzy]

Ahh yeah -- interesting.

Yep! I think the subdag thing is super powerful -- its kind of a different approach to dynamism. I've been comparing it to standard programming approaches:

• def <node>()... == variable assignment
• @parameterize == for loop
• @config.when == if
• @subdag == subroutine

And yep, agreed on warnings -- the execution component is highly pluggable, but most people will just be using out of the box threadpools or multiprocessing, so I think we need to be clear about it. And yeah -- so far we have some of these use-cases:

1. Over model parameters
2. Over chunks (files, groups of lines) of output stuff

A lot of the work we can do in the future is building higher-level interfaces for commonalities enabled by the more power-user features...

[jdonaldson]

Honestly, supporting dynamic model parameters is where I pulled up short with my own efforts. Dynamism is totally warranted there, because one may only want to know the parameters they wish to try once the current execution has finished. The interface may not be a map, but a fold over a DAG generator and some sort of dictionary/instance/value.

I guess it may make sense to do both... a "fully" dynamic operation where the results of the previous run fold into the next one, and a "parallelizable" operation where one can at least kick off a number of jobs simultaneously.

In the back of my mind I wanted to punt on this and find another framework that would do it for me (e.g. provide SLA and a nice monitoring application). Out of curiosity, did you have anything in that domain that caught your eye? A particular library or framework? It may not be something you want to own.

[elijahbenizzy]

So yeah, the initial design called for two types: Parallel[] and Sequential[], exactly as you described. Sequential[] is made for, say, mini-batch training, whereas Parallel[] is made for model parameters tuning or (more likely) processing like 10 files.

And I think that there are two axes here:

1. Dynamism (flexible # of nodes for a DAG, spawned at execution time)
2. Parallelism/execution (where to run these)

The power hamilton will provide is (1), and anything more complicated will be delegated to an external executor (this is highly pluggable in the dynamic version) -- which, ideally, should provide the task-management necessary to run/monitor. So that Hamilton just stores + updates the state. Note it doesn't (currently) persist execution state, so its a bit of a use-at-your-own-risk type thing. E.G. if you want something that runs, in a chron job, with a defined set of variables, and you get emailed if it fails, that's where airflow/prefect/metaflow/dagster comes in, and you run Hamilton within it. While we could go in that direction, its a crowded field, and the level above (micro-orchestration/code organization) is ripe for disruption.

That said, delegating just the parallel tasks is awkward, so having some capability in hamilton for dynamism that isn't too heavy-weight felt right. The implementation has the side-effect of executing nodes in groups, which actually makes delegating to a system such as ray or dask useful.

Re; other frameworks -- orchestration frameworks as mentioned + joblib/dask/ray/spark/other parallelism tooling are kind of what we're looking at. I think we want to keep the limit of a hamilton run as a single-task deal, otherwise you're breaking it up into components manually to run on orchestrators (or maybe using tooling we've been thinking about recently to generate the orchestration code).