So right now parallelizable needs the generator to be evaluated completely (it’s on the roadmap to remove that limitation). If I’m understanding correctly, in the interim, could you do:

# kick this off in some process
dr = driver.Builder().with_modules(...)...build() # create hamilton driver


batch = []
flush_interval = ...
last_flush = time.now()
while True:
   messages = queue.get(..., timeout=...)  # this waits for data from the queue
   if messages:
      batch += messages
   interval = time.now() - last_flush
   if len(batch) > ... or (batch and interval > flush_interval):
      if batch:
         result = dr.execute([...], inputs={"batch": batch})
         # do something with result
      last_flush = time.now()
      batch = []

> Can I wait for such events in a hamilton node? Thinking about it more. You could pass in a queue and hard block on it within a node.

def events(queue: ...) -> list[SomeEvent]:
   while True:
     # loop until event
     _events = queue.get(..)
     if _events:
        return ...

then the driver code is a simpler loop:

# kick this off in some process
dr = driver.Builder().with_modules(...)...build() # create hamilton driver


batch = []
flush_interval = ...
last_flush = time.now()
queue = ...
while True: # or some other loop condition
   results = dr.execute([...], inputs={"queue": queue}) # will only proceed when `events` would return 
   ...

Is it due to the queue / event ? or?

Hamilton nodes are serializable; the error here looks related to spawning something and it’s in the multiprocessing library

I have seen serialization issues before with inputs etc, but nothing like this error.

@Louwrens chatgpt suggests:

Avoid Passing Synchronization Primitives: Ensure that any object being passed between processes does not contain synchronization primitives. If you need synchronization between processes, consider using shared variables or

multiprocessing.Manager

to manage shared state instead.

I wanted to check with you guys. Now I'm thinking that using Redis might be an effective way to (a) send synchronization signals (via Redis publish/subscribe) to parallel hamilton nodes that the desired photo data is available in a python shared memory (the actual data will be a

numpy.ndarray

) and also (b) to pass the

shape

and

dtype

metadata of such shmem objects (via Redis set/get)... What do you think? I was thinking that if we're going to use Redis, we might use it as a caching backend for Hamilton too... Is that something realistic? or perhaps would you have alternatives that are already available to Hamilton that might do the signaling/metadata store features we're looking for?

I have seen serialization issues before with inputs etc, but nothing like this error.

Do you have examples around? perhaps there is a pattern that I can follow

Is it due to the queue / event ? or?

I believe the

multiprocessing.Event

object needs the right process context. Is there a way get this context from within the hamilton node? or perhaps, can I create a

context = multiprocessing.get_context()

myself and pass it to the Hamilton dynamic executor? so that I can attach my own

some_event = context.Event()

to it? ...

So big picture is that we have a robot that can pick up some object (we call them containers) and then takes photos of if, under an array of different modalities (object hold position, lighting conditions, camera choice, rotation index), these photos are to be analyzed with several calculations (ones using machine learning models and ones without) which results a number of defect reasons (or none) and the container is thereby classified as rejected or accepted (no (serious) defects). The goal is to efficiently classify many containers.

We have a lot of existing code in the robot that we cannot change easily. It triggers the photo taking loop and those photos are stored in shared memory objects (shmem) and we use Redis to store the metadata and to message any listeners when ready. We envision to spawn a process for each container to execute a Hamilton DAG that will do the analysis of defects. But also to do offline execution, when bunch of containers are loaded from disk, and a DAG can do parallel execution of all of them..

Within the DAG, some nodes will be loading the image data from shmem, or listen for Redis to message if they arent stored in shmem yet. These nodes are thus blocking -- perhaps until some timeout.

We see the options to run several things in parallel: • Containers: either via running the DAG in separate processes ourselves, or • Containers: within the DAG via

Parallelizable[]/Collect[]

• Analyses: within the DAG as distinct nodes • Analysis repetitions: some analyses have calculation nodes that are to be run for an array of photos/modalities (a handful), via

Parallelizable[]/Collect[]

or • Analysis repetitions: as distinct nodes, generate by

@parameterize

?

We have some questions: • Can we do

Parallelizable[]/Collect[]

for nodes that in between nodes that already do

Parallelizable[]/Collect[]

? • Can we do

async

nodes within synchronous nodes perhaps? • How can we do

@parameterize

(or similar) to generate a list of nodes that collect the desired photos from shmem, by iterating over several

Enum

names and/or repetition indices? (without manually writing out all the photo modalities in `dict`s)

We also want to do use a

GracefulErrorAdapter

for several kinds: • to abort all running container processes (major fault detected) • to abort a single container process (container can be rejected after early analysis already finds defects, without doing more analysis)

I'm hesitant to provide you with an actual graph that we have so far, because it is not sufficiently detailed of what we want and yet too revealing of the business.

e.g. could we suppose a graph of this kind:

A -> Bn         (A forks into Parallelizable n elements, say containers)
Bn -> Cnm       (another Parallelizable m elements, say repetitions)
Cnm -> helpers	(blocking calls to photoloader or calculation, possibly an `asyncio.gather()`?
Cnm -> Dn       (Collect m)
Dn -> E         (Collect n)

Probably we are working with a bunch of misconceptions about how to use Hamilton. Please let me know how I can better illustrate for you.

My last spam message, but not the least: Many thanks for writing such great library. We are very excited about it and very pleased with so many decisions you've made. gratitude thank you

> Can we do

Parallelizable[]/Collect[]

for nodes that in between nodes that already do

Parallelizable[]/Collect[]

? In the same DAG, not right now. We have not enabled that. But you could do Hamilton within Hamilton. E.g. a node that internally that executes its own Hamilton driver. > Can we do

async

nodes within synchronous nodes perhaps? Yes. Hamilton has an async driver, though it doesn’t support everything due to async. You could also have Hamilton within Hamilton, e.g. a node that creates a driver + asyncio.run > How can we do

@parameterize

(or similar) to generate a list of nodes that collect the desired photos from shmem, by iterating over several

Enum

names and/or repetition indices? (without manually writing out all the photo modalities in `dict`s) So to handle more dynamism it’s either parallelizable + collect, or resolving values at DAG start time (via @resolve_with_config), or you do it before creating the hamilton DAG and provide it as input (e.g. outerloop that delegates to a hamilton DAG, or do Hamilton within Hamilton). Let me know if that helps clarify things. My suggestion here is let’s work bottom up, design Hamilton DAGs for the specific components as you list out. Then we either manually orchestrate, or add Hamilton on top, or use something like Burr.