The AsyncDriver currently just works with the AsyncGraphAdapter. I would like it to work with a RayGraphAdapter. I am aware that Ray Tasks cannot themselves be async (if they wish to benefit from this then they need to start an async event loop inside the task) However I can see some benefit (not yet measurable, so I may be wrong!) in having all the coordination be async: HamiltonTracker, MLFlow, Ray task submission. Is this something that you are already considering? I am happy to take a look if not. Would your guidance be to extend the RayGraphAdapter to auto-detect when it is running in an Event loop or to subclass as AsyncRayGraphAdapter?

Just to understand better, what's the driver for async? Running in a webservice? Or you want to use some async libs? Or?

Not a webservice, no. Yes we do want to run async libs (Database queries, opening video frames from disk both have high I/O bottlenecks), however these will be within the Ray Tasks (ideally looking to use the Ray Object Store with zero-copy as far as possible to share data between tasks). The coordination layer also seems to have a lot of IO bottlenecks since it is talking to network services...now sure, we expect that the heavy compute in the Tasks will likely be taking by far the most time for the pipeline, I am interested to experiment with how much time can be shaved off by speeding up the coordination layer....it may only be useful for simpler development/PoC tasks, but these may be the ones where human time waiting for action matters most & so be worth it even there. As I say, I am very open to this turning out to be not as worth it as I hope 🙂

I think it would be easier to have a separate async path — you generally need to do that anyway since python doesn’t allow you to nest async within sync.. most libraries these days that try to support both have a

a*

version of a function/method

Another possibility is that some work could happen outside of a Ray node and thus be fully async, as well as avoiding the Ray overhead for smaller nodes....e.g. a DataLoader (if the coordination node is on the same host as the Ray workers then it can populate the Ray object store) I haven't yet looked at how we could do this within Hamilton...separate DAGs, subdags, etc

We could also make the ray graph adapter not

raify

everything — based on some tags …

That's a good idea to explore 🙂

So yeah, quite a few possibilities and design limitations. Might be nice to make this more concrete to help narrow down these ideas — any chance you could draw out a simple workflow/outline that would illustrate the benefit? Code/pseudocode (e.g. with your ideal API/assuming async + tag worked together. Would be easier to grok through some code I think (for me at least). Then it can also be a GitHub issue/let others comment!

Getting a bit wild brainstorming here, but thoughts on sending some tasks to Ray & some to e.g. Dask?

some work could happen outside of a Ray node and thus be fully async

to add one more thought: Hamilton within Hamilton. E.g. top level Hamilton handles Ray. Inside Ray tasks is async independent Hamilton async..

Getting a bit wild brainstorming here, but thoughts on sending some tasks to Ray & some to e.g. Dask?

entirely do-able — that’s just a different adapter…

At the moment I'm not sure what an ideal solution looks like...I'm trying to setup some options for ways to do stuff & then benchmark them to see which works well in practise for different usecases

> So we’d load up both Adapters. Or create some adapter that handles both… independent of them. Up to our imagination here. E.g. first version it duplicates their logic, second version, we inherit/override…

FWIW dynamic execution mode actually has the constructs for all of these things. Delegating to different executors — might be able to handle async as well with some cleverness (would have to experiment). But basically it has a construct that assigns executors to tasks, and you can easily switch between.

That sounds very cool 🙂 Flexibility like this allows for better optimisation

It was built for dynamic number of tasks (e.g. process 100 files), but will work for more static workflows

my pie in the sky is that we can learn to auto group nodes together to optimize for SERDE…. e.g. first run less optimal, but next run we try grouping more, etc… @Fran Boon but yeah as @Elijah Ben Izzy said. Some more requirements to help guide would be helpful.

@Fran Boon OK, some reading on dynamic execution: • Initial launch post — focuses on the use-case • More info on docs: https://hamilton.dagworks.io/en/latest/concepts/parallel-task/ Then the details (more documented by code) — note this is very into the weeds, but there’s a pretty powerful customizability system — really sharing this with you cause you’re likely one of a few who might happily leverage it 🙂 No worries if not, a lot of the same things you can do with the standard system. 1. Driver calls out to a graph_executor 2. If you enable v2 execution, it uses the task-based GraphExecutor (https://github.com/DAGWorks-Inc/hamilton/blob/d90212fc772ecc06689afd1aa43d5adc1d77cf73/hamilton/driver.py#L171) 3. The execute method is relatively clear: https://github.com/DAGWorks-Inc/hamilton/blob/d90212fc772ecc06689afd1aa43d5adc1d77cf73/hamilton/driver.py#L220 4. The key for you is two pieces (1) the grouping strategy, specified on with_grouping_strategy, as well as the execution manager, specified with with_execution_manager (execution manager bypasses the need for local/remote executors). 5. For a POC, you should be able to use GroupNodesIndividually as the grouping strategy, unless you’re doing something with parallelism 6. Then you can create a delegating execution manager that will delegate to one of the ray task executor, dask task executor, standard default one, or even create an async one — this async one could take an async task and run it against the passed in/current event loop (might need a small bit of surgery there). You could also get fancy enough to execute all of these against an event loop, thus bypassing a lot of the async/await magic, but it’s a bit tricky to think through that without writing some code. But yeah, whether this is the road you want to take without seeing the code you want to optimize/how you might approach it without hamilton

That is really interesting to go through...I really appreciate you taking the time to do this level of detail...it does look really promising, although none of that is yet Async, so that would need a fair bit of work to get through. I may start just with AsyncRayGraphAdapter & then adding a switch to execute nodes outside of Ray via a Tag...those both look very easy to implement & gives me a lot to play with already. If I do get to the stage of wanting to send some Tasks to Dask, then that's probably the time to really need the TaskBasedGraphExecutor

Makes sense! I think you could get it to work but definitely start from where you’re comfortable

@Fran Boon also, in case it's helpful, what's generally worked for us in these cases with lots of design possibilities, is to sketch out the user docs you'd want someone to follow. We find it helps steer the design to simpler implementations.