Hi All, is there an easy way to ensure that a function is always called (and always at the beginning of the DAG run) even if no requested final variable requires it? I can think of hacky solutions, but maybe there’s a “proper” way? (In case you wonder why: This would serve to initialize a module that the different DAG nodes themselves require. I usually just initialize from outside the DAG before running it. However, now I’m moving to a Ray runner and have to do the initialization step remotely.)

🤔 nothing first class. Which Ray integration are you using? I assume you don’t want those things to depend on a function? That would be the most straightforward and you would be able to encode it into the graph explicitly. Otherwise not thinking about it for very long, here are two ways that could work: 1. have your code import that module (either in the function or as a module import). So when deserialized it should be invoked. 2. maybe there’s an adapter that you could create that could check for initialization in the process it’s running in?

Thanks @Stefan Krawczyk. Indeed I didn’t want my functions to depend on a special node just when on Ray, but I might do that. I use RayGraphAdapter, what would it take to encode the initialization step there?

> Indeed I didn’t want my functions to depend on a special node just when on Ray, but I might do that. > > I use RayGraphAdapter, what would it take to encode the initialization step there? Okay what I said wouldn’t work. Because this adapter injects a

ray.remote

around every function - so it wouldn’t do what you need I think. 🤔 hmm. I think you could build a little adapter and implement the

run_before_node_execution

function, e.g. like the PrintLn adapter (see blog post on adapters), and have it do what you need. Make sure the adapter is serializable and then add it to the adapters list.

I need the step to only run once at the very beginning, so I suppose I will do that in

pre_graph_execute

then? At any rate this is more than enough to get me going, thanks!

So with the

RayGraphAdapter

each function will be serialize and deserialized and run as its own ray task. So I don’t understand “what” needs to be set up (e.g. is it local to the python process?) - it could be that you need to run it before each “node”. Or yeah, just use

pre_graph_execute

if it really only needs to be run once.

Yes, I guess I didn’t describe entirely the issue. But the short of it is, I have some code to execute in my first node (whichever it is) when running on Ray. So this solution looks good, executing essentially before the first node, I will test it right away.

To add

pre_graph_execute

is what you want if it’s idempotent and before every execution — if it’s just on driver instantiation, you’ll want

pre_do_anything

— https://github.com/DAGWorks-Inc/hamilton/blob/7f4bc6633bc683b511cf95dbab1521d01bc894af/hamilton/lifecycle/base.py#L263

I have tried and if I make a pre-graph adapter it doesn’t see to run at all when the backend is Ray, even on a local cluster. It runs fine in-thread without Ray. I will try to post some code here, maybe I’m missing something.

Oh interesting… Yeah let us know, that’s probably a bug? Oh wait, yeah, if it’s per python process ray might not propogate it

@Zoltan Eisler my understanding is that you want to instantiate something local to the python process (that’s why multithreading works)? If so, the way SERDE works would require you to put it into

run_before_node_execution

so that when the task runs in the ray task process, it puts whatever is needed into the python process that you need to run.

Exactly! I will put together a minimal test on a local cluster tomorrow.

yep. In case it’s helpful knowing about

__getstate__

&

__setstate__

(example) can help here as objects are serialized and deserialized. Notoriously DB connections etc are not serializeable

Ah cool, I’ll give this a try for fun!