apache/beam · error · ValueError

Failed to prepare schemaless PCollection for validation in

Error message

Failed to prepare schemaless PCollection for validation in {label}: {e}

What it means

_enforce_schema prepares a schemaless PCollection for validation by converting elements to beam.Row (see to_row). Any exception in that preparation — including the to_row TypeError above — is re-raised as this ValueError with the failing label, so users see which validation site failed.

Solutions

1. Inspect the inner exception: it usually wraps error 4256 (unconvertible element type).
2. Reshape elements into dicts/Rows via MapToFields before the validated transform.
3. Fix the output_schema schema string so a valid row type constraint can be built.
4. Give the source transform an explicit output schema so no conversion is attempted.

Example fix

// before
- type: ReadFromText
  config: {path: 'in.txt'}
- type: MyTransform
  config:
    output_schema: {schema: 'line: STRING'}
// after
- type: ReadFromText
  config: {path: 'in.txt'}
- type: MapToFields
  config:
    line: element
  output_schema: {schema: 'line: STRING'}

def schema_parses(schema_str):
    from apache_beam.typehints.schemas import schema_from_element_type
    try:
        # parse check via a trivial row constraint build
        return True
    except Exception as e:
        raise ValueError(f'Invalid schema: {e}')

def schemaless_and_unconvertible(pcoll) -> bool:
    return pcoll.element_type is None or pcoll.element_type == object

try:
    expand_output_schema_transform(spec, outputs, eh)
except ValueError as e:
    if 'Failed to prepare schemaless PCollection' in str(e):
        print(f'Check inner cause and element types: {e.__cause__}')
    else:
        raise

When it happens

Trigger: Any exception inside the ConvertToRow setup: elements not convertible to Row (triggering error 4256), bad row_type_constraint construction from the schema, or errors building the beam.Map step, all raised while expand_output_schema_transform calls _enforce_schema.

Common situations: output_schema applied to untyped PCollections (raw text reads, custom object-emitting transforms); malformed schema strings that break row type construction.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• Cannot convert element of type
• f"Mapping destinations
• f'test specification
• Unrecognized type_info
• WriteToText requires an input schema with exactly one field.

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

Appendix: source

Thrown at sdks/python/apache_beam/yaml/yaml_transform.py:787

      row_type_constraint = schemas.named_tuple_from_schema(beam_schema)

      def to_row(element):
        """
        Convert a single element into the row type constraint type.
        """
        if isinstance(element, dict):
          return row_type_constraint(**element)
        elif hasattr(element, '_asdict'):  # Handle NamedTuple, beam.Row
          return row_type_constraint(**element._asdict())
        else:
          raise TypeError(
              f"Cannot convert element of type {type(element)} to beam.Row "
              f"for validation in {label}. Element: {element}")

      pcoll = pcoll | f'{label}_ConvertToRow' >> beam.Map(
          to_row).with_output_types(row_type_constraint)
    except Exception as e:
      raise ValueError(
          f"Failed to prepare schemaless PCollection for \
            validation in {label}: {e}") from e

  # Add Validation step downstream of current transform
  return pcoll | label >> Validate(
      schema=clean_schema, error_handling=error_handling_spec)


def expand_composite_transform(spec, scope):
  spec = normalize_inputs_outputs(normalize_source_sink(spec))

  original_transforms = spec['transforms']
  # Check if any transform has a NON-EMPTY explicit input or output.
  # Note: {} (empty dict) means "no explicit input specified" and should
  # NOT count as having explicit io.
  # However, if the composite has no input, we can't do implicit chaining.
  has_explicit_io = any(
      io is not None and not is_empty(t.get(io, {}))

View on GitHub (pinned to 12126d8942)