apache/beam · error · TypeError

Unexpected output type: %s

Error message

Unexpected output type: %s

What it means

TypeError raised in AppliedPTransform.replace_output when the replacement output is not a PValue or dict of outputs. Only PValue instances (or dict mapping tags to them) are accepted when rewiring a transform's outputs after pipeline modification.

Source

Thrown at sdks/python/apache_beam/pipeline.py:1347

  def replace_output(
      self,
      output: Union[pvalue.PValue, pvalue.DoOutputsTuple],
      tag: Union[str, int, None] = None) -> None:
    """Replaces the output defined by the given tag with the given output.

    Args:
      output: replacement output
      tag: tag of the output to be replaced.
    """
    if isinstance(output, pvalue.DoOutputsTuple):
      self.replace_output(output[output._main_tag])
    elif isinstance(output, pvalue.PValue):
      self.outputs[tag] = output
    elif isinstance(output, dict):
      for output_tag, out in output.items():
        self.outputs[output_tag] = out
    else:
      raise TypeError("Unexpected output type: %s" % output)

    # Importing locally to prevent circular dependency issues.
    from apache_beam.transforms import external
    if isinstance(self.transform, external.ExternalTransform):
      self.transform.replace_named_outputs(self.named_outputs())

  def replace_inputs(self, main_inputs):
    self.main_inputs = main_inputs

    # Importing locally to prevent circular dependency issues.
    from apache_beam.transforms import external
    if isinstance(self.transform, external.ExternalTransform):
      self.transform.replace_named_inputs(self.named_inputs())

  def replace_side_inputs(self, side_inputs):
    self.side_inputs = side_inputs

    # Importing locally to prevent circular dependency issues.

View on GitHub (pinned to 12126d8942)

Solutions

1. Pass a PCollection (or other PValue) as the output
2. Pass a dict {tag: PCollection} to replace multiple outputs at once
3. Verify the value you are passing is the actual output of a transform, not intermediate data

Example fix

// before
applied.replace_output(None, results_list)
// after
applied.replace_output(None, new_pcollection)

from apache_beam.pvalue import PValue
assert isinstance(output, (PValue, dict)), f'Bad output: {output}'

from apache_beam.pvalue import PValue

def is_valid_output(x):
    return isinstance(x, PValue) or (isinstance(x, dict) and all(isinstance(v, PValue) for v in x.values()))

try:
    applied.replace_output(tag, output)
except TypeError as e:
    logging.error('replace_output requires a PValue: %s', e)

When it happens

Trigger: Calling applied_transform.replace_output(tag, something) with a plain Python value, list, or None instead of a PCollection.

Common situations: Programmatic pipeline surgery (pipeline.replace()/test scenarios) where users pass wrong objects; refactorings that changed outputs from PCollection to raw values.

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

• Delete passed string argument instead of list: %s
• schema_update_options must be a list. Received %s.
• Transform "{full_label}" was applied to the output of "{prod
• Only one of context or default_environment may be specified.
• Unknown annotation type %r (type %s) for %s

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