apache/beam · error · KeyError

Tag %r is not a defined output tag of

Error message

Tag %r is not a defined output tag of %s.

What it means

After running a pipeline with materialized outputs, results are accessed per output tag via the materialized result mapping. If you index with a tag that wasn't a defined output tag of the deferred PTransform, a KeyError with this message is raised.

Solutions

1. Index with a tag defined by the transform (main tag or one of deferred._tags).
2. Inspect the transform's output tags before indexing (e.g. deferred._tags).
3. Use the default indexing (no tag or MAIN_TAG) for single-output transforms.

Example fix

// before
result['output']  # tag doesn't exist
// after
result[beam.pvalue.TaggedOutput tag from transform]  # or result['main_tag'] as defined

available_tags = list(mat_result._deferred._tags)
assert tag in available_tags or tag == mat_result._deferred._main_tag, f"{tag!r} not in {available_tags}"

try:
  elements = mat_result[tag]
except KeyError:
  elements = mat_result[mat_result._deferred._main_tag]

When it happens

Trigger: result['wrong_tag'] on a _MaterializedResultBackedPCollectionView wrapper where the tag isn't in results_by_tag — e.g. misspelled tag or indexing a single-output transform with a made-up tag like 'out'.

Common situations: Multi-output transforms (DoFn with tagged outputs) where the developer indexes a tag not produced; copy-pasted tag names from other transforms.

Understand the failure class

Background: Record Not Found Errors: "not found", RecordNotFound, and "was not found" — what they mean and how to fix them — this error's family across 28 libraries.

Related errors

• A BigQuery table or a query must be specified
• A cluster_identifier should be Optional[Union[str…
• A context manager constructor (not a fully constructed…
• A has been supplied to the model handler, but the required…
• A pubsub message attribute key must not exceed 256 bytes.

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/ptransform.py:216

    self._result_id = result_id
    self.elements = []  # type: list[Any]

  def __reduce__(self):
    # When unpickled (during Runner API roundtrip serailization), get the
    # _MaterializedResult object from the cache so that values are written
    # to the original _MaterializedResult when run in eager mode.
    return (_get_materialized_result, (self._pipeline_id, self._result_id))


class _MaterializedDoOutputsTuple(pvalue.DoOutputsTuple):
  def __init__(self, deferred, results_by_tag):
    super().__init__(None, None, deferred._tags, deferred._main_tag)
    self._deferred = deferred
    self._results_by_tag = results_by_tag

  def __getitem__(self, tag):
    if tag not in self._results_by_tag:
      raise KeyError(
          'Tag %r is not a defined output tag of %s.' % (tag, self._deferred))
    return self._results_by_tag[tag].elements


class _AddMaterializationTransforms(_PValueishTransform):
  def _materialize_transform(self, pipeline):
    result = _allocate_materialized_result(pipeline)

    # Need to define _MaterializeValuesDoFn here to avoid circular
    # dependencies.
    from apache_beam import DoFn
    from apache_beam import ParDo

    class _MaterializeValuesDoFn(DoFn):
      def __init__(self):
        self.is_materialize_values_do_fn = True

      def process(self, element):

View on GitHub (pinned to 12126d8942)