Dear Heamilton Community, I am using Dynamic DAGs/Parallel Execution. What I am looking for is a fail tollerancy of the parallely executed tasks. I expect my processing to fail in unforseen ways and therefore don't want the process to stop entirely. Is there a way in the hamilton framework to archive that? An optimal behaviour would be for the failed node to logg its input, the error and the output from the prallelization method (i want to know what the intital source was since the input to a node propably doesn't have a clear identifier). Thanks for any hints

Hi @Richard Schulz! Do you have a picture of the DAG to share? (

Driver.display_all_functions()

) There are 2 approaches to tackle the problem that comes to mind: 1. Use

GracefulErrorAdapter

(reference) 2. Use the new caching feature (tutorial) Note that they can't be used together ## 1. GracefulErrorAdapter This is a lightweight adapter that gives "fault tolerance"

dr = (
    driver.Builder()
    .with_modules(my_module)
    .with_adapters(
        default.GracefulErrorAdapter(
            error_to_catch=KeyError,
            sentinel_value=None
        )
    )
    .build()
)

If the DAG execution encounters the exception specified in

error_to_catch

, the node will return the

sentinel_value

and continue execution. Setting

error_to_cache=BaseException

should catch all exceptions. This will ensure that

Driver.execute()

returns a result ## 2. Caching Caching will store results on disk after each node is executed.

dr = (
    driver.Builder()
    .with_modules(my_module)
    .with_cache()
    .build()
)

If a task fails midway through, its intermediary results will be stored. However, I think if any task fails, it will stop execution (not fault tolerance). You will be able to inspect the cached results and the

Driver

object will store structure logs (which can also be stored to file directly via

.with_cache(log_to_file=True)

). Also, if you're to run the DAG a second time, it will skip the successful branches by using cached values @Elijah Ben Izzy thoughts?

Thanks a lot that sound like usefull functions. I just made a small test setup to see how hamilton behaves on exceptions. So i dont have the real thing yet.

Great! Keep us in the loop!

Thanks @Thierry Jean — you’ve got it. @Richard Schulz — to add, here’s a fully working example. Note that the sum_process_foo takes in

None

for any failed task — you can add logging/whatnot here (or in the task). Caching + rerunning is a pretty good way to do retries if you’re interested, but there are a few other ways to approach that as well if that’s what you’re going for. That said, I think this will get at what you want. One trick — in this case we make the sentinel value

None

, as it should produce

None

when it fails and we can check it in the iterable. You could also make it

0

in this case, which would give it a default value! Then you don’t have to do the null check.

from hamilton import driver
from hamilton.ad_hoc_utils import create_temporary_module
from hamilton.htypes import Parallelizable, Collect
from hamilton.lifecycle import GracefulErrorAdapter


class MyException(Exception):
    pass


def foo() -> Parallelizable[int]:
    for i in range(10):
        yield i


def process_foo(foo: int) -> int:
    if foo % 2 == 0:
        raise MyException("foo is even")
    return foo + 1


def sum_process_foo(process_foo: Collect[int]) -> int:
    return sum((item for item in process_foo if item is not None))


if __name__ == "__main__":
    dr = (
        driver
        .Builder()
        .with_modules(create_temporary_module(foo, process_foo, sum_process_foo))
        .enable_dynamic_execution(allow_experimental_mode=True)
        .with_adapter(GracefulErrorAdapter(error_to_catch=MyException, sentinel_value=None))
        .build())
    print(dr.execute(["sum_process_foo"]))