Write the failures

To write the failures to a dead letter queue, convert them to a flat format with a documented schema:

Field (Java / Python)	Type	Description
`pipelineStep`	string, nullable	Name of the failing step
`inputElement`	string	Input element of the failing step
`originElement`	string, nullable	Origin element, when it’s tracked
`exceptionType`	string	Class name of the exception (the original one for a non serializable exception)
`exceptionMessage`	string, nullable	Message of the exception
`stackTrace`	string	Stack trace, causes included
`timestamp`	datetime, nullable	When the failure was created

BigQuery

FailureTransforms.toRows() converts the failures to Beam Rows with the FailureTransforms.SCHEMA, used by BigQueryIO to create the table:

result.failures()
        .apply("Failures to rows", FailureTransforms.toRows())
        .apply("Write failures", BigQueryIO.<Row>write()
                .to("my-project:my_dataset.failures")
                .useBeamSchema()
                .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
                .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));

FailureTransforms.toRow(failure) converts a single failure, e.g. in your own transform.

result.failures()
    .apply("Failures to rows", FailureTransforms.toRows())
    .apply("Write failures", BigQueryIO.write<Row>()
        .to("my-project:my_dataset.failures")
        .useBeamSchema()
        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND))

Failure.to_dict() converts a failure to a flat and JSON serializable dict, with the same fields in snake_case (pipeline_step, input_element…). FAILURE_BIGQUERY_SCHEMA is the matching BigQuery table schema:

from asgarde import FAILURE_BIGQUERY_SCHEMA, Failure

(result.failures
 | 'Failures to dicts' >> beam.Map(Failure.to_dict)
 | 'Write failures' >> beam.io.WriteToBigQuery(
     'my-project:my_dataset.failures',
     schema=FAILURE_BIGQUERY_SCHEMA,
     create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
     write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))

The Java and Python pipelines write to the same table format.