apache/beam · error · ValueError

Id ' ' is already taken.

Error message

Id '%s' is already taken.

What it means

PipelineContext.put_proto registers a proto under an id; by default, inserting an id that is already taken raises ValueError("Id '%s' is already taken."). This guards against silently overwriting pipeline components (transforms, pcollections) referenced by id.

Solutions

1. Pass ignore_duplicates=True when re-registering an identical proto under an existing id is acceptable
2. Generate a fresh unique id instead of reusing the existing one
3. Look up the existing component with context.get_by_id(id) instead of inserting again
4. Check why ids collide (e.g. pipeline construction or proto deserialization generating duplicate ids)

Example fix

// before
context.put_proto('transform_1', proto)  # raises if taken
// after
context.put_proto('transform_1', proto, ignore_duplicates=True)

if id in context._id_to_proto:
    id = context.next_id()  # or reuse get_by_id(id)

try:
    context.put_proto(id, proto)
except ValueError as e:
    if 'already taken' in str(e):
        context.put_proto(context.next_id(), proto)

When it happens

Trigger: Calling context.put_proto(id, proto) with an id already present in the context, e.g. re-adding a component during pipeline construction or re-hydrating a pipeline from proto with clashing ids, without ignore_duplicates=True.

Common situations: Custom runner/tooling code that manipulates PipelineContext directly; reconstructing pipelines from serialized protos where two components got the same id; calling get_by_proto (which auto-assigns via put_proto) for a proto already registered under a different id.

Related errors

• Blocking computation failed. State
• Cannot access OnTimerContext outside of @OnTimer methods.
• Cannot access OnWindowExpirationContext outside of…
• Cannot determine context class
• Cannot insert different protos %r and %r with the same ID %r

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/pipeline_context.py:143

          return id
    return self.put_proto(
        self._pipeline_context.component_id_map.get_or_assign(
            obj=obj, obj_type=self._obj_type, label=label),
        maybe_new_proto)

  def get_id_to_proto_map(self) -> dict[str, message.Message]:
    return self._id_to_proto

  def get_proto_from_id(self, id: str) -> message.Message:
    return self.get_id_to_proto_map()[id]

  def put_proto(
      self,
      id: str,
      proto: message.Message,
      ignore_duplicates: bool = False) -> str:
    if not ignore_duplicates and id in self._id_to_proto:
      raise ValueError("Id '%s' is already taken." % id)
    elif (ignore_duplicates and id in self._id_to_proto and
          self._id_to_proto[id] != proto):
      raise ValueError(
          'Cannot insert different protos %r and %r with the same ID %r',
          self._id_to_proto[id],
          proto,
          id)
    self._id_to_proto[id] = proto
    return id

  def __getitem__(self, id: str) -> Any:
    return self.get_by_id(id)

  def __contains__(self, id: str) -> bool:
    return id in self._id_to_proto


class PipelineContext(object):

View on GitHub (pinned to 12126d8942)