apache/beam · error · ValueError

No producer for %s

Error message

No producer for %s

What it means

Raised during Pipeline.from_runner_api when deserializing a pipeline proto that contains a PCollection with no producing transform. Every PCollection in a valid pipeline graph must be generated by some transform; Beam throws because the reconstructed graph would be incomplete.

Source

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

        })
    from apache_beam.runners import pipeline_context
    context = pipeline_context.PipelineContext(
        proto.components, requirements=proto.requirements)
    if proto.root_transform_ids:
      root_transform_id, = proto.root_transform_ids
      p.transforms_stack = [context.transforms.get_by_id(root_transform_id)]
    else:
      p.transforms_stack = [AppliedPTransform(None, None, '', None, None, None)]
    # TODO(robertwb): These are only needed to continue construction. Omit?
    p.applied_labels = {
        t.unique_name
        for t in proto.components.transforms.values()
    }
    for id in proto.components.pcollections:
      pcollection = context.pcollections.get_by_id(id)
      pcollection.pipeline = p
      if not pcollection.producer:
        raise ValueError('No producer for %s' % id)

    # Inject PBegin input where necessary.
    from apache_beam.io.iobase import Read
    from apache_beam.transforms.core import Create
    has_pbegin = [Read, Create]
    for id in proto.components.transforms:
      transform = context.transforms.get_by_id(id)
      if not transform.inputs and transform.transform.__class__ in has_pbegin:
        transform.main_inputs = {'None': pvalue.PBegin(p)}

    if return_context:
      return p, context  # type: ignore  # too complicated for now
    else:
      return p


class PipelineVisitor(object):
  """For internal use only; no backwards-compatibility guarantees.

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure the proto's transforms include the transform whose output id matches the PCollection id
2. Regenerate the pipeline proto from a working pipeline rather than hand-editing it
3. Validate the serialized pipeline JSON references consistent pcollection/transform ids

for pc_id in proto.components.pcollections:
    assert pc_id in set().union(*[set(t.outputs) for t in proto.components.transforms.values()]), f'No producer for {pc_id}'

try:
    p = Pipeline.from_runner_api(proto, runner, options)
except ValueError as e:
    logging.error('Corrupt pipeline proto: %s', e)

When it happens

Trigger: Loading a hand-edited, truncated, or corrupted pipeline JSON/proto; constructing runner-api protos manually and forgetting to register the producing transform; pipeline fragments saved without their upstream transforms.

Common situations: Custom runners manipulating pipeline protos; cross-language pipelines where component protos are assembled programmatically and an id is mistyped.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• Unknown timing constant: " + timing
• Unknown PaneInfo encoding 0x" + encoding.toString(16)
• Unknown type tag %x
• Expected 0 or 1, got %s
• Could not find enum descriptor: {full_name}

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