Hello dear hamilton community, I have a question regarding the processing of large amounts of pictures. The processing pipeline which i want to migrate to hamilton processes one picture. Is it possible to have some kind of iterator in the hamilton ecosystem where one picture is pulled, the DAG runs and this is repeated until all pictures have been processed. Maybe rerun anytime there are new pictures in the folder/in the DB?

A hamilton construct for iteration could be

Parallelizable

and

Collect

where hamilton will run your DAG on nodes in between

Parallelizable/Collect

see here.

# image_processor.py
from typing import List, Any, Generator
from pathlib import Path

from hamilton.htypes import Parallelizable, Collect


def image_file(files_to_process: List[Path]) -> Parallelizable[Path]:
    for img in files_to_process:
        yield img


def processed_image(image_file: str) -> Any:
    # do something with image_file

    ...


def processed_images(processed_image: Collect[Any]) -> Any:
    # do something with processed_image which is a list
    ...

This would process each file asynchronously within hamilton dag. Below is some driver code to execute the image processing pipeline.

# run.py

import time
import image_processor
from hamilton import driver, executors

dr = (
    driver.Builder()
    .with_modules(image_processor)
    .enable_dynamic_execution(allow_experimental_mode=True)
    .with_local_executor(executors.SynchronousLocalTaskExecutor())
    .build()
)

I do not know if

hamilton

has an internal construct to re-run/re-trigger, but since you already have a pipeline, you could do this outside of

hamilton

.

while True:
    files_to_process = _watch_folder_for_new_files()
    if files_to_process:
        result = dr.execute(["processed_images"], inputs=dict(files_to_process=files_to_process))
    else:
        print("No new files to process.")
    time.sleep(10)  # Sleep for a while before checking for new files again

So you could execute your process on each image file without having to loop over all image files out side of hamilton and using the driver to execute inside each loop. The DAG would re-execute only if new files are present.

Thank you for the answer. It looks like I can make use of it. To clarify, my project is not in production phase but building a POC. Therefore I don't have any orcestration for the DAG yet. I just want every file to be processed independently by a DAG to be able to monitor at which steps fails are happening. Anyway the production remarks are also usefull for later.