Beam native error handling

Beam has a native error handling: some transforms and IOs (e.g. KafkaIO, BigQueryIO) can send their bad records to an ErrorHandler, as BadRecord objects. Asgarde and the Beam error handling work together: the Asgarde failures can be added to the same ErrorHandler, for a single dead letter queue for the whole pipeline.

When to use what

Need	Use
Errors of your own transformation steps (map, flatMap, filter, DoFn)	Asgarde: the failures of all the steps in a fluent flow, without boilerplate
Bad records of a Beam IO or transform supporting the ErrorHandler	The Beam ErrorHandler
Both in the same pipeline	Asgarde, with its failures converted to BadRecords and added to the Beam ErrorHandler

Add the Asgarde failures to a Beam ErrorHandler

Java

FailureTransforms.toBadRecords() converts the failures to Beam BadRecords:

final BadRecordErrorHandler<?> errorHandler = pipeline.registerBadRecordErrorHandler(new WriteBadRecordsToBigQuery());

// Bad records of a Beam IO supporting the error handler.
pipeline.apply("Read Kafka", KafkaIO.<String, String>read()
        // ...
        .withBadRecordErrorHandler(errorHandler));

// Failures of the Asgarde steps, added to the same error handler.
final WithFailures.Result<PCollection<Order>, Failure> result = CollectionComposer.of(messages)
        .apply("Parse", MapElementFn.into(TypeDescriptor.of(Order.class)).via(OrderParser::parse))
        .getResult();

errorHandler.addErrorCollection(result.failures().apply("To bad records", FailureTransforms.toBadRecords()));

errorHandler.close();

The mapping of a Failure to a BadRecord:

BadRecord	From the Failure
record.humanReadableJsonRecord	The origin element when it’s tracked, the input element otherwise
record.encodedRecord	Empty: Asgarde keeps the elements as strings
failure.exception	The exception, as a string
failure.exceptionStacktrace	The stack trace
failure.description	The pipeline step

Python

Failure.to_bad_record converts a failure to the dead letter format of the Beam with_exception_handling, (element, (exception type, exception repr, stack trace lines)). The element is the origin element when it’s tracked, the input element otherwise:

from apache_beam.transforms.error_handling import ErrorHandler

from asgarde import CollectionComposer, Failure

with ErrorHandler(beam.io.WriteToText('gs://my-bucket/dead-letters/bad-record')) as error_handler:
    # Bad records of a Beam transform using the error handler.
    parsed = messages | 'Parse' >> beam.Map(parse).with_exception_handling(error_handler=error_handler)

    # Failures of the Asgarde steps, added to the same error handler.
    result = CollectionComposer.of(parsed).map('Enrich', enrich).filter('Validate', is_valid)
    error_handler.add_error_pcollection(result.failures | 'To bad records' >> beam.Map(Failure.to_bad_record))

Any Beam transform in the composer

apply adds any Beam ParDo (beam.Map, beam.FlatMap, beam.ParDo(MyDoFn())) or DoFn to the flow, with the Beam native exception handling: its errors become Failure objects, gathered with the other steps.

result = (CollectionComposer.of(messages)
          .map('Decode', lambda message: message.decode())
          .apply('Parse', beam.ParDo(ParseOrderDoFn()))
          .filter('Validate', is_valid))

Beam only gives the representation of the exception: the failure has the original exception type, and the message of the last line of the stack trace. apply is not available while the origin element is tracked.

Note

The Asgarde Failure gives more than a BadRecord (e.g. the input element and the origin element separately, the failure metrics per step): keep the Asgarde failures as they are if you don’t use the Beam ErrorHandler.