Asgarde

Error handling and dead letter queues for Apache Beam Python, without the boilerplate. Also available for Java and Kotlin with Asgarde Java.

📖 Documentation: https://tosun-si.github.io/asgarde/

Why Asgarde

With plain Beam, each step needs its own DoFn with a try/except block and tagged outputs, and all the failures must be flattened at the end. Asgarde keeps a fluent flow and gathers the failures of all the steps:

from asgarde import CollectionComposer

result = (CollectionComposer.of(values)
          .map('Trim', lambda value: value.strip())
          .map('Parse', int)
          .filter('Keep even numbers', lambda number: number % 2 == 0))

outputs = result.outputs    # Output of the last step
failures = result.failures  # The failures of all the steps, for your dead letter queue

Installation

Asgarde is published on PyPI. apache-beam is a minimum version, not a pin: your pipeline brings its own Beam version.

uv add asgarde      # or: pip install asgarde

Features

• One place for all the errors: each step catches its errors in a Failure (with the stack trace), the CollectionComposer gathers the failures of all the steps.
• map, flat_map and filter operators with side inputs and DoFn lifecycle actions, and Failure.from_exception for your own DoFn.
• Origin element: with with_origin_element, the failures also give the element that entered the flow, to debug and replay from the start. Evaluated only when a failure occurs.
• Never breaks your job: non picklable exceptions, non JSON dicts, partial flat_map outputs, deterministic labels for Dataflow updates.
• Failure metrics: a Beam counter per step.
• Typed (py.typed).

Compatibility

Python 3.10 to 3.14, apache-beam>=2.60.0. The CI tests the minimum Beam version and the latest Beam release on every push and every week. See Compatibility.

Roadmap

See the roadmap: replayable input element, Beam schema for the failures, ready-to-use failure sinks...

Contributing

Contributions are welcome, see CONTRIBUTING.md.

License

MIT