Hi all, I’m using Hamilton for some file processing. The process involves splitting a PDF into images. Then, for each image, preprocessing, running Tesseract for OCR, and producing a summary. Requesting some feedback re: • Is my implementation reasonable or are there obvious anti-patterns? • Is this a good application for a subdag for each image (vs. the collapsed

processed_image

node)? • I don’t love some of my node names and abstractions —

each_image

,

processed_image

->

processed_images

😐. Any recommendations? ◦ This expansion is especially weird and points to poor naming:

list(_processed_image["preprocessed_image"] for _processed_image in processed_image)

I’ve collapsed the image processing steps into one node,

processed_image

. I’d rather explode this into

preprocessed_image

,

tesseract_output

,

tesseract_summary

, etc… Ideally, the results of each of these stages could be accessed without further processing before all results are Collected. With that in mind, I realize the current implementation is probably fine for this level of the problem. However, I’ll need a different solution for the “meta-scale” problem — processing thousands of PDFs. Within this context, I’ll need to queue, monitor, and save artifacts for each PDF individually. I wont be collapsing all processing into a single node as I’ve done here. I suspect this is where Hamilton’s scope ends and a full-featured Orchestrator is preferred. Is this correct? DAG graph is attached. DAG definition below.

# dag.py
from typing import Union
from pathlib import Path
from PIL import Image
import pandas as pd
from hamilton.function_modifiers import source, extract_fields
# These are excluded from the DAG
from utils import (
    pdf_to_images,
    preprocess_image,
    image_to_tesseract_data,
    TesseractDocumentSummary,
)
from hamilton.htypes import Parallelizable, Collect
import numpy as np


@source
def pdf_path() -> Union[str, Path]:
    pass


def pdf_images(pdf_path: Union[str, Path], kwargs: dict = {}) -> list[Image.Image]:
    default_kwargs = {"dpi": 300, "thread_count": 4}
    kwargs = {**default_kwargs, **kwargs}
    return pdf_to_images(pdf_path, **kwargs)


def each_image(pdf_images: list[Image.Image]) -> Parallelizable[Image.Image]:
    for image in pdf_images:
        yield image


def processed_image(each_image: Image.Image) -> dict:
    preprocessed_image = preprocess_image(each_image)
    tesseract_output = image_to_tesseract_data(preprocessed_image)
    difficulty_summary = TesseractDocumentSummary.from_tesseract_data(tesseract_output)

    return {
        "preprocessed_image": preprocessed_image,
        "tesseract_output": tesseract_output,
        "difficulty_summary": difficulty_summary,
    }


@extract_fields(
    {
        "preprocessed_images": list[np.ndarray],
        "tesseract_outputs": pd.DataFrame,
        "difficulty_table": pd.DataFrame,
    }
)
def processed_images(
    processed_image: Collect[dict],
) -> dict:
    return {
        "preprocessed_images": list(
            _processed_image["preprocessed_image"]
            for _processed_image in processed_image
        ),
        "tesseract_outputs": pd.concat(
            [
                _processed_image["tesseract_output"]
                for _processed_image in processed_image
            ],
            ignore_index=True,
        ),
        "difficulty_table": pd.DataFrame(
            [
                _processed_image["difficulty_summary"]
                for _processed_image in processed_image
            ]
        ),
    }

Hey. Make sense: 1. The implementation makes sense. I can see why you decouple the list of images from the yielding, although I’d recommend putting those in one. 2. Nothing looks crazy aobut how you’re doing it — there’s a bit of dimensional cleverness (E.G. contacenating DFs/whatnot), but I wouldn’t think you’re doing the wrong thing a. You could do a single node for all the processing, but this gives you less visibility b. If you did a single node it could manage parallelism itself — E.g. run a bunch of subdags within a single node c. Depends how much control you want to give the framework 3. I don’t hate them, but this is one of the few problems with this API — the plurals don’t match up. I usually juust call it something like “input_image: Collect[…]” -> “processed_image”, but it’s not too far off 4. Regarding splitting — you could split it out and then join them together before the collect a. You can run each DAG individaully for testing b. I’m not 100 % sure what you mean by “accessed without further processing before all results are Collected — mind expanding? Otherwise, sure you can use an orchestrator and break Hamilton up into multiple DAGs with a queue. That said, thousands, IMO, is not meta-scale. If you have billions its a serious eng problem — hamilton should happily be able to handle thousands or millions IMO. The key here, however, is knowing what is stored in memory on the executor and what is stored in memory on the nodes: 1. All inputs to the subdag (parallel outputs) and outputs (collected stuff) is held in memory in a list 2. All intermediate stuff is stored in memory at the place that it is being executed. Also,

Parallelizable

and

Collect

have a ton of customizability that can probably help you with your execution problems At a high-level, you’ll want to implement (or use a prebuilt) TaskExecutor. Then you’ll want to call with_remote_executor and pass an instance in. The cool part about this is that you can bridge with any other execution system you want — E.G. a fancy one with queueing, etc… For instance, if you have batched inputs on ray at high throughput using the GPU, all you’d have to do is implement the submit_task method to submit it to a queue that gives you a future. This also allows you to interact with any orchestrator/executor technology with a programmatic python API. All that said, my guess is that something like Ray for execution. See this blog post for more. The only caveat is that I’d make the subdags (between parallel and collect) reseponsible for loading up the image data, processing it, then saving it out, and passing pointers. That way the central orchestrator is working with pointers, rather than large in-memory objects. S3 or a NFS is a simple solution here.

+1 to what @Elijah Ben Izzy said. I'll also mention that

processed_image

could internally be calling another Hamilton DAG; this is effectively what subdag (see the docs) is syntactic sugar for.

for naming convention, I like to highlight the singular / plural in node names and the collection. I edited the code a bit to give you an idea. A key idea here is that "inside"

Parallelizable/Collect

you can split operations in several nodes to process "one document at a time". Then, your

Collect

node can collect multiple "inside nodes". You can't have multiple

Collect

nodes though, i.e., having

processed_image_collection

and

tessteract_output_collection

would be invalid

from hamilton.function_modifiers import extract_fields
from hamilton.htypes import Collect, Parallelizable
import pandas as pd
import numpy as np

def pdf_image() -> Parallelizable[np]:
    default_kwargs = {"dpi": 300, "thread_count": 4}
    kwargs = {**default_kwargs, **kwargs}
    for image in pdf_to_images(pdf_path, **kwargs):
      yield image
    
def preprocessed_image(pdf_image: str) -> np.ndarray:
    return preprocess_image(each_image)

def tesseract_output(preprocessed_image: np.ndarray) -> str:
    return image_to_tesseract_data(preprocessed_image)

def difficulty_summary(tesseract_output: str) -> float:
    return TesseractDocumentSummary.from_tesseract_data(tesseract_output)


@extract_fields(
    {
        "preprocessed_images": list[np.ndarray],
        "tesseract_outputs": dict,
        "difficulty_table": dict,
    }
)
def pdf_image_collection(
    preprocessed_image: Collect[np.ndarray],
    tesseract_output: Collect[str],
    difficulty_summary: Collect[float],
) -> dict:
    return {
        "preprocessed_images": preprocessed_image,  # is a list by default
        "tesseract_outputs": pd.concat(tesseract_output, ignore_index=True),
        "difficulty_table": pd.DataFrame(difficulty_summary),
    }
    
    
import functions
from hamilton import driver, base

dr = ( 
  driver.Builder()
  .with_modules(functions)
  .build()
)

dr.display_all_functions(orient="TB", show_legend=False)