apache/beam · error · RuntimeError

Please specify InteractiveRunner when creating the Beam pipe

Error message

Please specify InteractiveRunner when creating the Beam pipeline to use this function on unbouded PCollections.

What it means

pipeline_fragment tooling builds a sub-pipeline fragment and instruments it. Before running, it inspects the instrumented pipeline proto: if any PCollection in the fragment is UNBOUNDED it raises this RuntimeError, because fragment execution requires bounded PCollections — unbounded ones need a full InteractiveRunner-driven pipeline that supports streaming recording. (The message has a typo: 'unbouded'.)

Source

Thrown at sdks/python/apache_beam/runners/interactive/pipeline_fragment.py:130

    from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
    try:
      if isinstance(self._runner_pipeline.runner, InteractiveRunner):
        preserved_skip_display = self._runner_pipeline.runner._skip_display
        preserved_force_compute = self._runner_pipeline.runner._force_compute
        preserved_blocking = self._runner_pipeline.runner._blocking
        self._runner_pipeline.runner._skip_display = not display_pipeline_graph
        self._runner_pipeline.runner._force_compute = not use_cache
        self._runner_pipeline.runner._blocking = blocking
        return fragment.run()
      else:
        pipeline_instrument = instr.build_pipeline_instrument(
            fragment, self._runner_pipeline._options)
        pipeline_instrument_proto = (
            pipeline_instrument.instrumented_pipeline_proto())
        if any(pcoll.is_bounded == beam_runner_api_pb2.IsBounded.UNBOUNDED
               for pcoll in
               pipeline_instrument_proto.components.pcollections.values()):
          raise RuntimeError(
              'Please specify InteractiveRunner when creating '
              'the Beam pipeline to use this function '
              'on unbouded PCollections.')
        result = beam.pipeline.Pipeline.from_runner_api(
            pipeline_instrument_proto, fragment.runner,
            fragment._options).run()
        result.wait_until_finish()
        ie.current_env().mark_pcollection_computed(
            pipeline_instrument.cached_pcolls)
        return result
    finally:
      if isinstance(self._runner_pipeline.runner, InteractiveRunner):
        self._runner_pipeline.runner._skip_display = preserved_skip_display
        self._runner_pipeline.runner._force_compute = preserved_force_compute
        self._runner_pipeline.runner._blocking = preserved_blocking

  def _build_runner_pipeline(self):
    runner_pipeline = beam.pipeline.Pipeline.from_runner_api(

View on GitHub (pinned to 12126d8942)

Solutions

1. Create the pipeline with InteractiveRunner: beam.Pipeline(runner=InteractiveRunner()) so unbounded PCollections are handled by the interactive machinery.
2. Replace or bound the unbounded source for fragment analysis (e.g. a bounded Read or a finite Generate).
3. For streaming, run the full interactive pipeline (ib.show/ib.record with windowing) rather than the fragment path.
4. Isolate the unbounded branch away from the fragment being inspected — the check triggers if ANY pcoll in the fragment is unbounded.
5. Check source transforms for boundedness before fragment analysis.

Example fix

// before: p = beam.Pipeline(); pc = p | ReadFromPubSub(...) ; PipelineFragment([pc]).run() | // after: p = beam.Pipeline(runner=InteractiveRunner()); pc = p | ReadFromPubSub(...); ib.show(pc)

from apache_beam.runners.interactive import interactive_runner; def is_interactive_pipeline(p): return isinstance(p.runner, interactive_runner.InteractiveRunner); assert is_interactive_pipeline(pipeline)  # before fragment analysis of streaming pcolls

def fragment_is_bounded(pcolls): unbounded_markers = ('ReadFromPubSub', 'ReadFromKafka', 'TestStream'); return not any(m in str(p) for p in pcolls for m in unbounded_markers)

try: fragment_result = PipelineFragment([pcoll]).run() | except RuntimeError as e: (pipeline = beam.Pipeline(runner=InteractiveRunner()); rebuild pcoll and use ib.show) if ('unbouded' in str(e) or 'InteractiveRunner' in str(e)) else raise

When it happens

Trigger: Using PipelineFragment / fragment-based display paths on a pipeline containing an infinite beam.Generate, an unbounded ReadFromPubSub/Kafka, a TestStream, or any transform producing unbounded PCollections while the pipeline was created with a plain runner (e.g. DirectRunner) instead of InteractiveRunner.

Common situations: Streaming notebooks: users experimenting with Pub/Sub or Kafka sources then invoking fragment inspection/show tooling; pipelines built before switching to interactive mode; TestStream-based tests opened interactively.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• To write a GlobalWindow unbounded PCollection, triggering_fr
• UnboundedSource restriction was neither finished nor checkpo
• Watermark must be monotonically increasing.Provided watermar
• MLTransform only supports GlobalWindows when producing artif
• Transform node %r was not replaced as expected.

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