apache/beam · error · ValueError

The input to this transform does not appear to be an error…

Error message

The input to this transform does not appear to be an error output.  Expected a schema'd input with a field named 'msg' or 'message' or 'exception'

What it means

StripErrorMetadata extracts the original failing element from an error output PCollection. It requires the input to be schema'd and contain one of the recognized error-record fields ('failed_row', 'element', 'record' per the source; the message text mentions 'msg'/'message'/'exception' variants). If neither holds, it raises this ValueError.

Solutions

1. Point StripErrorMetadata's input at the transform's error output, e.g. `input: MyTransform.errors`, not the main output.
2. Ensure the upstream error output is schema'd and carries the failing element in a field named 'failed_row', 'element', or 'record'.
3. If using a custom error format, normalize it via MapToFields to produce one of the expected field names first.

Example fix

# before
- type: StripErrorMetadata
  input: MyMapTransform   # main output, not errors
# after
- type: StripErrorMetadata
  input: MyMapTransform.errors

from apache_beam.typehints import schema_from_element_type
fields = {f.name for f in schema_from_element_type(pcoll.element_type).fields}
assert fields & {'failed_row', 'element', 'record'}, 'not an error output pcoll'

def is_error_output(pcoll) -> bool:
    try:
        names = {f.name for f in schema_from_element_type(pcoll.element_type).fields}
    except TypeError:
        return False
    return bool(names & {'failed_row', 'element', 'record'})

try:
    stripped = pcoll | StripErrorMetadata()
except ValueError as e:
    if 'does not appear to be an error output' in str(e):
        raise ConfigError('wire StripErrorMetadata to <transform>.errors') from e

When it happens

Trigger: Wiring StripErrorMetadata (or using with_error_handling workflows) to a PCollection that is not an error output — e.g. the main output of a transform, a custom error output with differently-named fields, or an untyped PCollection.

Common situations: Connecting to the wrong output name (main output instead of `<transform>.errors`), using a hand-rolled error output whose payload field is named something other than failed_row/element/record, or a runner/transform version whose error schema uses different field names.

Related errors

• Can only use expressions on a schema'd input.
• violates schema
• Unknown grouping columns
• Unknown or unsupported atomic type
• Unrecognized type_info

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/116543b79f7e2ebf. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/yaml/yaml_mapping.py:492

  _ERROR_FIELD_NAMES = ('failed_row', 'element', 'record')

  def __init__(self):
    super().__init__(label=None)

  def expand(self, pcoll):
    try:
      existing_fields = {
          fld.name: fld.type
          for fld in schema_from_element_type(pcoll.element_type).fields
      }
    except TypeError:
      fld = None
    else:
      for fld in self._ERROR_FIELD_NAMES:
        if fld in existing_fields:
          break
      else:
        raise ValueError(
            'The input to this transform does not appear to be an error ' +
            "output.  Expected a schema'd input with a field named " +
            ' or '.join(repr(fld) for fld in self._ERROR_FIELD_NAMES))

    if fld is None:
      # This handles with_exception_handling() that returns bare tuples.
      return pcoll | beam.Map(lambda x: x[0])
    else:
      return pcoll | beam.Map(lambda x: getattr(x, fld)).with_output_types(
          typing_from_runner_api(existing_fields[fld]))


class Validate(beam.PTransform):
  """Validates each element of a PCollection against a json schema.

  Args:
      schema: A json schema against which to validate each element.
      error_handling: Whether and how to handle errors during iteration.

View on GitHub (pinned to 12126d8942)