apache/beam · error · RuntimeError

Streaming Python direct runner does not support…

Error message

Streaming Python direct runner does not support cross-language pipelines.Please use other runners such as FlinkRunner, DataflowRunner, or PrismRunner.

What it means

The streaming DirectRunner rejects cross-language pipelines: its visitor raises RuntimeError when it encounters an ExternalTransform (xlang transform) in the pipeline graph. Cross-language transforms need runner-side expansion services that the Python direct streaming path doesn't support, so the failure happens up front with suggested alternative runners.

Solutions

Switch to a runner that supports cross-language pipelines: FlinkRunner, DataflowRunner, or PrismRunner.
Test non-xlang parts of the pipeline locally and exercise xlang IO only on a supporting runner.
Remove the ExternalTransform for local tests, mocking its input/output.

Example fix

// before
# streaming xlang pipeline locally
with beam.Pipeline(runner='DirectRunner', options=streaming_opts) as p:
  p | KafkaIO.read(...)   # ExternalTransform -> RuntimeError
// after
with beam.Pipeline(runner='FlinkRunner', options=flink_opts) as p:
  p | KafkaIO.read(...)

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam.transforms.external import ExternalTransform
def uses_xlang(pipeline) -> bool:
    found = []
    pipeline.apply_visitor(lambda t: found.append(t) if isinstance(getattr(t, 'transform', None), ExternalTransform) else None)
    return bool(found)

Try / catch

try:
    result = pipeline.run()
except RuntimeError as e:
    if 'cross-language' in str(e):
        sys.exit('Use FlinkRunner/DataflowRunner/PrismRunner for xlang pipelines')
    raise

Prevention

Reserve DirectRunner for pure-Python batch pipelines.
Test cross-language IO only on runners with expansion support.
Centralize runner selection in config so local runs can be swapped easily.

When it happens

Trigger: Running a streaming pipeline on the DirectRunner that contains cross-language transforms such as Kafka IO via Java expansion (ExternalTransform).

Common situations: Prototyping a Kafka/Kinesis/xlang pipeline locally with --streaming before deploying to a real runner; tests that use cross-language IO with the direct runner.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/direct/direct_runner.py:570

  def run_pipeline(self, pipeline, options):
    """Execute the entire pipeline and returns an DirectPipelineResult."""

    # TODO: Move imports to top. Pipeline <-> Runner dependency cause problems
    # with resolving imports when they are at top.
    # pylint: disable=wrong-import-position
    from apache_beam.pipeline import PipelineVisitor
    from apache_beam.runners.direct.consumer_tracking_pipeline_visitor import ConsumerTrackingPipelineVisitor
    from apache_beam.runners.direct.evaluation_context import EvaluationContext
    from apache_beam.runners.direct.executor import Executor
    from apache_beam.runners.direct.transform_evaluator import TransformEvaluatorRegistry
    from apache_beam.testing.test_stream import TestStream
    from apache_beam.transforms.external import ExternalTransform

    class VerifyNoCrossLanguageTransforms(PipelineVisitor):
      """Visitor determining whether a Pipeline uses a TestStream."""
      def visit_transform(self, applied_ptransform):
        if isinstance(applied_ptransform.transform, ExternalTransform):
          raise RuntimeError(
              "Streaming Python direct runner "
              "does not support cross-language pipelines."
              "Please use other runners such as FlinkRunner, "
              "DataflowRunner, or PrismRunner.")

    pipeline.visit(VerifyNoCrossLanguageTransforms())

    # If the TestStream I/O is used, use a mock test clock.
    class TestStreamUsageVisitor(PipelineVisitor):
      """Visitor determining whether a Pipeline uses a TestStream."""
      def __init__(self):
        self.uses_test_stream = False

      def visit_transform(self, applied_ptransform):
        if isinstance(applied_ptransform.transform, TestStream):
          self.uses_test_stream = True

    visitor = TestStreamUsageVisitor()

View on GitHub (pinned to 12126d8942)