apache/beam · error · ValueError

'Ambiguous unnamed transform

Error message

'Ambiguous unnamed transform {transform["type"]}'

What it means

Raised by create_test's get_name helper when generating a test from a pipeline spec: multiple transforms share the same 'type' and none has an explicit 'name', so a stable generated name cannot be assigned. get_name falls back to the transform type only when it is unique in the spec; with two or more unnamed transforms of the same type, the name would be ambiguous, so construction aborts.

Solutions

1. Add a unique 'name' field to each transform of that type in the pipeline YAML
2. Rename one of the duplicate-type transforms so unnamed fallback naming is unambiguous

When it happens

Trigger: Thrown at sdks/python/apache_beam/yaml/yaml_testing.py:425 when the library encounters an invalid state.

Common situations: See trigger scenarios.

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

Appendix: source

Thrown at sdks/python/apache_beam/yaml/yaml_testing.py:425

    options = beam.options.pipeline_options.PipelineOptions(
        pickle_library='cloudpickle',
        **yaml_transform.SafeLineLoader.strip_metadata(
            pipeline_spec.get('options', {})))

  providers = yaml_provider.merge_providers(
      yaml_provider.parse_providers('', pipeline_spec.get('providers', [])),
      {
          'AssertEqualAndRecord': yaml_provider.as_provider_list(
              'AssertEqualAndRecord', AssertEqualAndRecord)
      })

  def get_name(transform):
    if 'name' in transform:
      return str(transform['name'])
    else:
      if sum(1 for t in transform_spec['transforms']
             if t['type'] == transform['type']) > 1:
        raise ValueError('Ambiguous unnamed transform {transform["type"]}')
      return str(transform['type'])

  input_transforms = [
      t for t in transform_spec['transforms'] if t['type'] != 'Create' and
      not yaml_transform.empty_if_explicitly_empty(t.get('input', []))
  ]

  mock_outputs = [{
      'name': get_name(t),
      'elements': [
          _try_row_as_dict(row)
          for row in _first_n(t, options, max_num_inputs, providers)
      ],
  } for t in input_transforms]

  output_transforms = [
      t for t in transform_spec['transforms'] if t['type'] == 'LogForTesting' or
      yaml_transform.empty_if_explicitly_empty(t.get('output', [])) or

View on GitHub (pinned to 12126d8942)