apache/beam · error · ValueError

Exploding unknown field

Error message

Exploding unknown field "{field}"

What it means

Beam YAML's Explode transform raises this when a field listed in the `explode` option does not exist in the input PCollection's element schema. It fails fast at pipeline construction time in `ExpandExplode.expand()` because exploding a non-existent field would silently produce wrong output. The field list is validated against `named_fields_from_element_type(pcoll.element_type)`.

Solutions

1. Fix the field name in the `explode` config to match a field that exists in the input schema.
2. Inspect the input PCollection schema (e.g. with `pcoll.element_type` or `beam.schema` logging) and confirm the field exists before the Explode step.
3. If the field is removed upstream, move the Explode before the removal or remove it from the explode list.
4. If the input lacks a schema, add one (e.g. via a Cast or ToRow step) before exploding.

Example fix

// before
- type: Explode
  input: rows
  config:
    explode: [tags, categoroies]
// after
- type: Explode
  input: rows
  config:
    explode: [tags, categories]

from apache_beam.typehints import row_type
fields_in_explode = {'tags', 'categories'}
schema_fields = {name for name, _ in __import__('apache_beam.yaml.yaml_mapping', fromlist=['named_fields_from_element_type']).named_fields_from_element_type(pcoll.element_type)}
unknown = fields_in_explode - set(schema_fields)
if unknown:
    raise ValueError(f'Explode fields not in input schema: {unknown}')

try:
    expanded = Explode(fields=['tags']).expand(pcoll)
except ValueError as e:
    if 'Exploding unknown field' in str(e):
        logger.error('Fix explode config: %s', e)
    raise

When it happens

Trigger: Calling the Explode transform (yaml_mapping.ExpandExplode) with self._fields containing a field name not present in the input PCollections's schema; typically via YAML `transform: Explode` with an `explode: [field]` config where `field` is misspelled, removed by an earlier step, or the input is not schema'd.

Common situations: Typo in field name in the YAML spec; referencing a field produced downstream instead of upstream; the input is a plain dict/row without a Beam schema so `named_fields_from_element_type` returns nothing; renaming fields in an earlier Map step without updating the explode list.

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

• Input schema has multiple fields
• Redefinition of field
• Can only append fields on a schema'd input.
• Can only drop fields if append is true.
• Can only drop fields on a schema'd input.

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

Appendix: source

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

      else:
        # Doesn't matter.
        cross_product = True
    self._fields = fields
    self._cross_product = cross_product
    # TODO(yaml):
    # 1. Support standard error handling argument.
    # 2. Supposedly error_handling parameter is not an accepted parameter when
    #    executing.  Needs further investigation.
    self._exception_handling_args = exception_handling_args(error_handling)

  @maybe_with_exception_handling
  def expand(self, pcoll):
    all_fields = [
        x for x, _ in named_fields_from_element_type(pcoll.element_type)
    ]
    for field in self._fields:
      if field not in all_fields:
        raise ValueError(f'Exploding unknown field "{field}"')
    to_explode = self._fields

    def explode_cross_product(base, fields):
      if fields:
        copy = dict(base)
        for value in base[fields[0]]:
          copy[fields[0]] = value
          yield from explode_cross_product(copy, fields[1:])
      else:
        yield beam.Row(**base)

    def explode_zip(base, fields):
      to_zip = [base[field] for field in fields]
      copy = dict(base)
      for values in itertools.zip_longest(*to_zip, fillvalue=None):
        for ix, field in enumerate(fields):
          copy[field] = values[ix]
        yield beam.Row(**copy)

View on GitHub (pinned to 12126d8942)