Failure

Each error caught in a step gives a Failure object, sent to the failures PCollection.

Java
Python

public class Failure implements Serializable {
    private final String pipelineStep;   // Name of the step
    private final String inputElement;   // Input element, as a string (toString)
    private final Throwable exception;   // Exception raised by the step
    private final String originElement;  // Element that entered the flow, null if not tracked (see Origin element)
    private final Instant timestamp;     // When the failure was created (since 1.3.0)

    public static <T> Failure from(String pipelineStep, WithFailures.ExceptionElement<T> exceptionElement);
    public static <T> Failure from(String pipelineStep, T element, Throwable exception);

    // Computed from the exception (since 1.3.0), e.g. to write the failure to a dead letter queue.
    public String getExceptionType();      // Class name, the original one for a SerializableThrowable
    public String getExceptionMessage();
    public String getStackTrace();         // Causes included

    // Getters...
}

To write the failures to BigQuery or to add them to a Beam ErrorHandler, see Write the failures and Beam native error handling.

If the exception can’t be serialized, it’s replaced by a SerializableThrowable keeping the original class name, message, stack trace and causes (see Failure handling guarantees).

@dataclass
class Failure:
    pipeline_step: str       # Name of the step
    input_element: str       # Input element, as a string (JSON for a dict)
    exception: Exception     # Exception raised by the step (always picklable)
    stack_trace: str = ''    # Stack trace, as a string
    origin_element: str | None = None  # Element that entered the flow (see Origin element)

    @classmethod
    def from_exception(cls, pipeline_step: str, element: Any, exception: Exception) -> 'Failure': ...

If the exception can’t be pickled, it’s replaced by a SerializableException keeping the original type and message. The stack trace is kept as a string because a pickled exception loses its traceback.

Input element format

The input element is stored as a string:

Java: the toString() of the element.
Python: the JSON string of a dict (non JSON types like datetime or bytes converted with str), the str() of any other element.

To control the format, implement the string representation of your objects, e.g. as JSON:

Java
Python

public class Team implements Serializable {
    // Fields...

    @Override
    public String toString() {
        // Jackson is already a dependency of Beam.
        return JsonUtil.serialize(this);
    }
}

@dataclass
class Team:
    name: str

    def __str__(self) -> str:
        return json.dumps(dataclasses.asdict(self))