apache/beam · error · ValueError

No producer for

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.

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

• Could not find enum descriptor
• Expected 0 or 1, got
• Faled loading session: expected dict, got
• from_runner_api_parameter does not handle empty or None…
• Invalid tag %r

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

Appendix: 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)