apache/beam · error · ValueError

f'Ambiguous output at line

Error message

f'Ambiguous output at line {SafeLineLoader.get_line(name)}: {name} has outputs {list(outputs.keys())}'

What it means

Scope.get_pcollection, when given a bare transform name (no `.output` suffix), can pick a PCollection automatically only if the transform has exactly one output — or exactly two where one is a configured error-handling output. Otherwise the choice is ambiguous and Beam raises this ValueError listing all output tags.

Solutions

Reference the specific output with a dotted name: `MyTransform.<tag>` using a tag from the error message.
Check the listed output tags in the message and pick the intended one.
If only the main output is wanted and exactly one error output exists, the auto-disambiguation already handles it — verify the error_handling.output config.
Restructure the pipeline so each consumer names its input output explicitly.

Example fix

# before
- name: consume
  input: split_and_validate
  type: WriteToJson
# after
- name: consume
  input: split_and_validate.valid_rows
  type: WriteToJson

Defensive patterns

Strategy: validation

Validate before calling

def check_reference(name, transform):
    n_outputs = len(transform.get('outputs', []))
    if n_outputs > 1 and '.' not in name:
        err_out = transform.get('config', {}).get('error_handling', {}).get('output')
        if not (err_out and n_outputs == 2):
            raise ValueError(f'{name} is multi-output; use name.tag')

Type guard

def needs_explicit_output(transform_spec):
    return len(transform_spec.get('outputs', [])) > 2

Try / catch

try:
    pcoll = scope.get_pcollection(name)
except ValueError as e:
    if 'Ambiguous output' in str(e):
        raise UserPipelineError(f'{name}: specify output tag, e.g. {name}.<tag>') from e

Prevention

Always use dotted references for multi-output transforms
Remember error_handling adds a second output
Prefer explicit output tags over bare names in generated pipelines

When it happens

Trigger: Using a bare `input: MyTransform` reference where the transform produces multiple outputs (e.g. a partition-style transform or any transform with error_handling configured) without specifying which output tag to consume.

Common situations: Multi-output transforms feeding a downstream transform without a dotted tag; forgetting that adding error_handling made a previously single-output transform ambiguous; generated references that omit tags.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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

Appendix: source

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

        return outputs[output]
      elif len(outputs) == 1 and outputs[next(iter(outputs))].tag == output:
        return outputs[next(iter(outputs))]
      else:
        raise ValueError(
            f'Unknown output {repr(output)} '
            f'at line {SafeLineLoader.get_line(name)}: '
            f'{transform} only has outputs {list(outputs.keys())}')
    else:
      outputs = self.get_outputs(name)
      if len(outputs) == 1:
        return only_element(outputs.values())
      else:
        error_output = self._transforms_by_uuid[self.get_transform_id(
            name)]['config'].get('error_handling', {}).get('output')
        if error_output and error_output in outputs and len(outputs) == 2:
          return next(
              output for tag, output in outputs.items() if tag != error_output)
        raise ValueError(
            f'Ambiguous output at line {SafeLineLoader.get_line(name)}: '
            f'{name} has outputs {list(outputs.keys())}')

  def get_outputs(self, transform_name):
    return self.compute_outputs(self.get_transform_id(transform_name))

  @memoize_method
  def compute_outputs(self, transform_id):
    return expand_transform(self._transforms_by_uuid[transform_id], self)

  def best_provider(
      self, t, input_providers: yaml_provider.Iterable[yaml_provider.Provider]):
    if isinstance(t, dict):
      spec = t
    else:
      spec = self._transforms_by_uuid[self.get_transform_id(t)]
    possible_providers = []
    unavailable_provider_messages = []

View on GitHub (pinned to 12126d8942)