apache/beam · error · ValueError

"%s" requires a pipeline to be specified as there are no def

Error message

"%s" requires a pipeline to be specified as there are no deferred inputs.

What it means

When applying a PTransform via | on values that have no deferred inputs (no PCollections), Beam needs a pipeline context to build the transform. If no pipeline was passed, self.pipeline is None, and no pipelines were inferred, ValueError is raised asking for an explicit pipeline.

Source

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

    """Used to apply this PTransform to non-PValues, e.g., a tuple."""
    pvalueish, pvalues = self._extract_input_pvalues(left)
    if isinstance(pvalues, dict):
      pvalues = tuple(pvalues.values())
    pipelines = [v.pipeline for v in pvalues if isinstance(v, pvalue.PValue)]
    if pvalues and not pipelines:
      deferred = False
      # pylint: disable=wrong-import-order, wrong-import-position
      from apache_beam import pipeline
      from apache_beam.options.pipeline_options import PipelineOptions

      # pylint: enable=wrong-import-order, wrong-import-position
      p = pipeline.Pipeline('DirectRunner', PipelineOptions(sys.argv))
    else:
      if not pipelines:
        if self.pipeline is not None:
          p = self.pipeline
        else:
          raise ValueError(
              '"%s" requires a pipeline to be specified '
              'as there are no deferred inputs.' % self.label)
      else:
        p = self.pipeline or pipelines[0]
        for pp in pipelines:
          if p != pp:
            raise ValueError(
                'Mixing values in different pipelines is not allowed.'
                '\n{%r} != {%r}' % (p, pp))
      deferred = not getattr(p.runner, 'is_eager', False)
    # pylint: disable=wrong-import-order, wrong-import-position
    from apache_beam.transforms.core import Create

    # pylint: enable=wrong-import-order, wrong-import-position
    replacements = {
        id(v): p | 'CreatePInput%s' % ix >> Create(v, reshuffle=False)
        for (ix, v) in enumerate(pvalues)
        if not isinstance(v, pvalue.PValue) and v is not None

View on GitHub (pinned to 12126d8942)

Solutions

Apply the transform within a `with beam.Pipeline() as p:` context and pass pipeline-aware inputs (e.g. p | transform).
Pass the pipeline explicitly (transform.with_pipeline(p) or apply via pipeline.apply).
Set the transform's pipeline attribute before application if used standalone.
Use beam.Create inside a pipeline scope for constant inputs.

Example fix

// before
result = beam.Map(lambda x: x + 1) | 5  # no pipeline
// after
with beam.Pipeline() as p:
  result = p | beam.Create([5]) | beam.Map(lambda x: x + 1)

Defensive patterns

Strategy: try-catch

Validate before calling

assert 'pipeline' in dir() or pipeline is not None, "provide a pipeline when applying transform to non-deferred input"

Try / catch

try:
  result = transform | value
except ValueError as e:
  if 'requires a pipeline' in str(e):
    with beam.Pipeline() as p:
      result = p | beam.Create([value]) | transform
  else:
    raise

Prevention

Always apply transforms inside `with beam.Pipeline()` scope
Use beam.Create for plain Python values

When it happens

Trigger: Applying a transform to a plain value (e.g. beam.Create-like usage via __ror__ with non-PCollection inputs) without a pipeline argument and without self.pipeline set.

Common situations: Calling ptransform | value outside a with-beam.Pipeline block; using eager standalone transforms without supplying a Pipeline.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

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