Asgarde

Java

final WithFailures.Result<PCollection<Integer>, Failure> result = CollectionComposer.of(words)
        .apply("Map", MapElements.into(TypeDescriptors.strings()).via((String word) -> word + "Test"))
        .apply("Filter", FilterFn.by(word -> word.length() > 3))
        .apply("Word length", MapElementFn.into(TypeDescriptors.integers()).via(word -> 1 / word.length()))
        .getResult();

final PCollection<Integer> outputs = result.output();
final PCollection<Failure> failures = result.failures(); // All the failures of all the steps

Kotlin

val result: Result<PCollection<Int>, Failure> = CollectionComposer.of(words)
    .map("Map") { word -> word + "Test" }
    .filter("Filter") { word -> word.length > 3 }
    .mapFn("Word length", { word -> 1 / word.length })
    .result

Python

result = (CollectionComposer.of(words)
          .map('Map', lambda word: f'{word}Test')
          .filter('Filter', lambda word: len(word) > 3)
          .map('Word length', lambda word: 1 / len(word)))

outputs = result.outputs
failures = result.failures  # All the failures of all the steps

One place for all the errors

Each step catches its errors in a Failure object. The composer gathers the failures of all the steps in a single PCollection, ready for your dead letter queue.

Keeps the fluent style

Chain your transforms as usual: no tuple tags, no try/catch, no Flatten of the failures in your code.

Never breaks your job

Non serializable exceptions, failing toString, partial flatMap outputs, unstable transform names: the error handling itself can’t make the job fail.

Not tied to a Beam version

Your pipeline brings its own Beam version. The CI tests every week against the latest Beam release.