apache/beam · error · ValueError

Mixing values in different pipelines is not allowed.

Error message

Mixing values in different pipelines is not allowed.
{%r} != {%r}

What it means

When a PTransform is applied to multiple inputs, all input PCollections must belong to the same Pipeline object. __ror__ compares collected pipelines and raises ValueError if any differ, because mixing PCollections across pipelines is undefined in Beam.

Solutions

Create all inputs within the same Pipeline instance.
Re-read the source data under the target pipeline instead of reusing a PCollection from another pipeline.
Restructure code so a single pipeline owns every PCollection being combined.

Example fix

// before
p1 = beam.Pipeline(); p2 = beam.Pipeline()
a = p1 | 'A' >> beam.Create([1])
b = p2 | 'B' >> beam.Create([2])
combined = a | beam.Flatten(passthrough=b)  # mixes pipelines
// after
p = beam.Pipeline()
a = p | 'A' >> beam.Create([1])
b = p | 'B' >> beam.Create([2])
combined = (a, b) | beam.Flatten()

Defensive patterns

Strategy: validation

Validate before calling

pipelines = {getattr(v, 'pipeline', None) for v in inputs if hasattr(v, 'pipeline')}
assert len(pipelines) <= 1, f"inputs span multiple pipelines: {pipelines}"

Try / catch

try:
  combined = a | flatten_b ...
except ValueError as e:
  if 'Mixing values in different pipelines' in str(e):
    # rebuild all inputs under one pipeline
    raise
  raise

Prevention

Create every PCollection under the same Pipeline object
Avoid caching PCollections across pipeline lifetimes in tests/scripts

When it happens

Trigger: Combining PCollections from two different `beam.Pipeline()` instances (e.g. p1 | Join(p2_read_value)) or reusing a PCollection created under an old pipeline in a new one.

Common situations: Building two pipelines in a script/REPL and later trying to union/join them; iterating pipelines in tests and mixing fixtures from different pipelines.

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

Appendix: source

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

      # 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
    }
    pvalueish = _SetInputPValues().visit(pvalueish, replacements)
    self.pipeline = p
    result = p.apply(self, pvalueish, label)
    if deferred:
      return result
    _allocate_materialized_pipeline(p)

View on GitHub (pinned to 12126d8942)