As a data engineer, you’re likely familiar with the challenges of error handling in Apache Beam Java applications. Traditional approaches can lead to verbose code, making it difficult to read and maintain. The Asgarde library offers a solution by providing a way to write less code and produce more concise and expressive code.

What is Asgarde?

Asgarde is an open-source library that simplifies error handling in Apache Beam Java applications. It accomplishes this by wrapping common error handling patterns into reusable components. This can save you time and effort when writing Beam pipelines, and it can also make your code easier to read and understand.

Benefits of Using Asgarde

There are several benefits to using Asgarde in your Beam applications:

• Reduced boilerplate code: Asgarde eliminates the need to write repetitive error handling code. This can significantly improve the readability and maintainability of your pipelines.
• Improved error handling: Asgarde provides a centralized location for handling errors, which can make it easier to debug and troubleshoot your pipelines.
• More concise code: Asgarde’s reusable components can help you write more concise code that is easier to understand.

Key Features of Asgarde

Asgarde offers a number of features that can simplify error handling in your Beam applications, including:

• Centralized error handling: Asgarde provides a centralized location for handling errors in your pipelines. This can make it easier to debug and troubleshoot errors.
• Exception handling: Asgarde provides a way to handle exceptions in a more concise and expressive way.
• DoFn lifecycle methods: Asgarde provides lifecycle methods for DoFn that can be used to handle errors in a more robust way.

Comparison between approaches

Usual Beam pipeline with error handling

WithFailures.Result<PCollection<String>, Failure> result1 = input
        .apply("Map", MapElements
                        .into(TypeDescriptors.strings())
                        .via((String word) -> word + "Test")
                        .exceptionsInto(TypeDescriptor.of(Failure.class))
                        .exceptionsVia(exElt -> Failure.from("step", exElt)));

final PCollection<String> output1 = result1.output();
final PCollection<Failure> failure1 = result1.failures();

WithFailures.Result<PCollection<String>, Failure> result2 = output1
        .apply("FlatMap", FlatMapElements
                            .into(TypeDescriptors.strings())
                            .via((String line) -> Arrays.asList(Arrays.copyOfRange(line.split(" "), 1, 5)))
                            .exceptionsInto(TypeDescriptor.of(Failure.class))
                            .exceptionsVia(exElt -> Failure.from("step", exElt)));

final PCollection<String> output2 = result1.output();
final PCollection<Failure> failure2 = result1.failures();

final PCollectionTuple result3 = output2
        .apply("Map ParDo", ParDo.of(wordCountFn).withOutputTags(wordCountFn.getOutputTag(), TupleTagList.of(wordCountFn.getFailuresTag())));

final PCollection<Integer> output3 = result3.get(wordCountFn.getOutputTag());
final PCollection<Failure> failure3 = result3.get(wordCountFn.getFailuresTag());

final PCollection<Failure> allFailures = PCollectionList
        .of(failure1)
        .and(failure2)
        .and(failure3)
        .apply(Flatten.pCollections());

Beam pipeline with Asgarde

final WithFailures.Result<PCollection<Integer>, Failure> resultComposer = CollectionComposer.of(input)
        .apply("Map", MapElements.into(TypeDescriptors.strings()).via((String word) -> word + "Test"))
        .apply("FlatMap", FlatMapElements
                .into(TypeDescriptors.strings())
                .via((String line) -> Arrays.asList(Arrays.copyOfRange(line.split(" "), 1, 5))))
        .apply("ParDo", MapElementFn.into(TypeDescriptors.integers()).via(word -> 1 / word.length()))
        .getResult();

Conclusion

Asgarde is a valuable tool that can simplify error handling in Apache Beam Java applications. By using Asgarde, you can write less code, produce more concise and expressive code, and improve the overall maintainability of your Beam pipelines.

Reference

The Asgarde library is available on GitHub: https://github.com/tosun-si/asgarde

If you’re looking for a way to simplify error handling in your Beam applications, I encourage you to check out Asgarde.

• Java
• Beam