{"record":{"id":"705e37d65531e48b","repo":"apache/beam","slug":"please-specify-interactiverunner-when-creating-the-beam","errorCode":null,"errorMessage":"Please specify InteractiveRunner when creating the Beam pipeline to use this function on unbouded PCollections.","messagePattern":"Please specify InteractiveRunner when creating the Beam pipeline to use this function on unbouded PCollections\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/interactive/pipeline_fragment.py","lineNumber":130,"sourceCode":"    from apache_beam.runners.interactive.interactive_runner import InteractiveRunner\n    try:\n      if isinstance(self._runner_pipeline.runner, InteractiveRunner):\n        preserved_skip_display = self._runner_pipeline.runner._skip_display\n        preserved_force_compute = self._runner_pipeline.runner._force_compute\n        preserved_blocking = self._runner_pipeline.runner._blocking\n        self._runner_pipeline.runner._skip_display = not display_pipeline_graph\n        self._runner_pipeline.runner._force_compute = not use_cache\n        self._runner_pipeline.runner._blocking = blocking\n        return fragment.run()\n      else:\n        pipeline_instrument = instr.build_pipeline_instrument(\n            fragment, self._runner_pipeline._options)\n        pipeline_instrument_proto = (\n            pipeline_instrument.instrumented_pipeline_proto())\n        if any(pcoll.is_bounded == beam_runner_api_pb2.IsBounded.UNBOUNDED\n               for pcoll in\n               pipeline_instrument_proto.components.pcollections.values()):\n          raise RuntimeError(\n              'Please specify InteractiveRunner when creating '\n              'the Beam pipeline to use this function '\n              'on unbouded PCollections.')\n        result = beam.pipeline.Pipeline.from_runner_api(\n            pipeline_instrument_proto, fragment.runner,\n            fragment._options).run()\n        result.wait_until_finish()\n        ie.current_env().mark_pcollection_computed(\n            pipeline_instrument.cached_pcolls)\n        return result\n    finally:\n      if isinstance(self._runner_pipeline.runner, InteractiveRunner):\n        self._runner_pipeline.runner._skip_display = preserved_skip_display\n        self._runner_pipeline.runner._force_compute = preserved_force_compute\n        self._runner_pipeline.runner._blocking = preserved_blocking\n\n  def _build_runner_pipeline(self):\n    runner_pipeline = beam.pipeline.Pipeline.from_runner_api(","sourceCodeStart":112,"sourceCodeEnd":148,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/interactive/pipeline_fragment.py#L112-L148","documentation":"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'.)","triggerScenarios":"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.","commonSituations":"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.","solutions":["Create the pipeline with InteractiveRunner: beam.Pipeline(runner=InteractiveRunner()) so unbounded PCollections are handled by the interactive machinery.","Replace or bound the unbounded source for fragment analysis (e.g. a bounded Read or a finite Generate).","For streaming, run the full interactive pipeline (ib.show/ib.record with windowing) rather than the fragment path.","Isolate the unbounded branch away from the fragment being inspected — the check triggers if ANY pcoll in the fragment is unbounded.","Check source transforms for boundedness before fragment analysis."],"exampleFix":"// before: p = beam.Pipeline(); pc = p | ReadFromPubSub(...) ; PipelineFragment([pc]).run() | // after: p = beam.Pipeline(runner=InteractiveRunner()); pc = p | ReadFromPubSub(...); ib.show(pc)","handlingStrategy":"type-guard","validationCode":"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","typeGuard":"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)","tryCatchPattern":"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","preventionTips":["Use InteractiveRunner whenever a pipeline may contain unbounded sources.","Keep streaming (Pub/Sub/Kafka/infinite Generate) pcolls out of fragment inspection.","For streaming, use ib.show/ib.record with windowing instead of fragment tooling.","Check source transforms for boundedness before fragment analysis."],"tags":["python","apache-beam","interactive","streaming","runtime-error"],"backgroundTag":"unsupported-operation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}