apache/beam · error · TypeError

A cluster_identifier should be Optional[Union[str, beam.Pipe

Error message

A cluster_identifier should be Optional[Union[str, beam.Pipeline, ClusterMetadata], instead %s was given.

What it means

Interactive Beam's cluster manager helper (_cluster_metadata) resolves a cluster_identifier into a ClusterMetadata. It only accepts a cluster name string, a beam.Pipeline, or an existing ClusterMetadata; anything else is rejected with a TypeError raised from the DataprocClusterManager machinery in interactive_beam.py. Note: the raise passes the message and argument as tuple positionals, so the printed message ends with a raw tuple rather than a formatted string, but the intent is clear: the cluster_identifier you passed is of an unsupported type.

Source

Thrown at sdks/python/apache_beam/runners/interactive/interactive_beam.py:594

      elif isinstance(cluster_identifier, ClusterMetadata):
        meta = cluster_identifier
        if meta in self.dataproc_cluster_managers:
          meta = self.dataproc_cluster_managers[meta].cluster_metadata
        elif (meta and self.default_cluster_metadata and
              meta.cluster_name == self.default_cluster_metadata.cluster_name):
          _LOGGER.warning(
              'Cannot change the configuration of the running cluster %s. '
              'Existing is %s, desired is %s.',
              self.default_cluster_metadata.cluster_name,
              self.default_cluster_metadata,
              meta)
          meta.reset_name()
          _LOGGER.warning(
              'To avoid conflict, issuing a new cluster name %s '
              'for a new cluster.',
              meta.cluster_name)
      else:
        raise TypeError(
            'A cluster_identifier should be Optional[Union[str, '
            'beam.Pipeline, ClusterMetadata], instead %s was given.',
            type(cluster_identifier))
    return meta

  def _cleanup(self, dcm: DataprocClusterManager) -> None:
    dcm.cleanup()
    self.dataproc_cluster_managers.pop(dcm.cluster_metadata, None)
    self.master_urls.pop(dcm.cluster_metadata.master_url, None)
    for p in dcm.pipelines:
      self.pipelines.pop(p, None)
    if dcm.cluster_metadata == self.default_cluster_metadata:
      self.default_cluster_metadata = None


# Users can set options to guide how Interactive Beam works.
# Examples:
# ib.options.enable_recording_replay = False/True

View on GitHub (pinned to 12126d8942)

Solutions

Pass a ClusterMetadata object obtained from ib.cluster_metadata instead of a raw handle.
Pass the beam.Pipeline object itself (pipeline = beam.Pipeline(); ...; ib.cluster_metadata(pipeline)).
Pass the cluster name string exactly as returned by the cluster manager (meta.cluster_name).
If you must accept arbitrary input, coerce to one of the three supported types before calling.
Read the printed tuple in the traceback to see the actual type you passed (the second tuple element is type(cluster_identifier)) and fix the variable's source.

Example fix

// before: result = pipeline.run(); ib.cluster_metadata(result)  # PipelineResult, not accepted | // after: ib.cluster_metadata(pipeline)  # pass the beam.Pipeline itself, a str name, or a ClusterMetadata

Defensive patterns

Strategy: type-guard

Validate before calling

import apache_beam as beam; from apache_beam.runners.interactive.dataproc.types import ClusterMetadata; def is_valid_cluster_identifier(x): return x is None or isinstance(x, (str, beam.Pipeline, ClusterMetadata)); assert is_valid_cluster_identifier(ident)

Type guard

def as_cluster_identifier(x): import apache_beam as beam; from apache_beam.runners.interactive.dataproc.types import ClusterMetadata; return x if x is None or isinstance(x, (str, beam.Pipeline, ClusterMetadata)) else (_ for _ in ()).throw(TypeError(f'Unsupported cluster_identifier type: {type(x)}'))

Try / catch

try: ib.cluster_metadata(ident) | except TypeError as e: logger.error('Bad cluster_identifier %r: %s', ident, e); ident = ib.cluster_metadata(pipeline)  # fall back to pipeline handle

Prevention

Only pass values returned by ib.cluster_metadata, the pipeline object, or meta.cluster_name.
Normalize notebook UI inputs to one of the three supported types at the call site.
Remember PipelineResult is NOT accepted — keep the pipeline object around.
Check type(cluster_identifier) before calling when wiring wrappers.

When it happens

Trigger: Calling ib.cluster_metadata (and related evict/show_cluster_into paths) with a cluster_identifier that is not a str, not a beam.Pipeline, and not a ClusterMetadata — e.g. passing an int pipeline id, a PipelineResult, a dict of options, or a wrapper object holding the pipeline.

Common situations: Notebook users wiring up interactive Dataproc workflows: storing a pipeline handle in a wrapper and passing the wrapper; passing the result of pipeline.run() (a PipelineResult) instead of the pipeline; typing a cluster id into a variable holding bytes or a numpy str_; older notebooks written against an API version that accepted other identifier shapes.

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/df185cd77e8fb3e1. Report an issue: GitHub.