apache/beam · error · NotImplementedError

You must specify a --job_endpoint when using…

Error message

You must specify a --job_endpoint when using --runner=PortableRunner. Alternatively, you may specify which portable runner you intend to use, such as --runner=FlinkRunner or --runner=SparkRunner.

What it means

PortableRunner has no built-in job server; it needs a --job_endpoint pointing at an already-running job service. default_job_server() raises this NotImplementedError because the base PortableRunner cannot start a server itself, unlike concrete runners (FlinkRunner, SparkRunner) that override it.

Solutions

Start a job service and pass --job_endpoint=<host:port> (e.g. localhost:8099).
Use a concrete portable runner instead: --runner=FlinkRunner or --runner=SparkRunner.
If you only want local execution, use --runner=DirectRunner rather than PortableRunner.

Example fix

// before
--runner=PortableRunner
// after
--runner=PortableRunner --job_endpoint=localhost:8099

Defensive patterns

Strategy: validation

Validate before calling

opts = PipelineOptions(flags)
portable = opts.view_as(PortableOptions)
if portable.runner_type == 'PortableRunner' and not portable.job_endpoint:
    raise SystemExit('--runner=PortableRunner requires --job_endpoint (or use FlinkRunner/SparkRunner)')

Try / catch

try:
    result = pipeline.run()
    result.wait_until_finish()
except NotImplementedError as e:
    if 'job_endpoint' in str(e):
        sys.exit('Set --job_endpoint or choose FlinkRunner/SparkRunner')
    raise

Prevention

Never select PortableRunner without also setting --job_endpoint.
Prefer concrete runners (FlinkRunner, SparkRunner) which supply default job servers.
Validate pipeline options in your launcher script before submitting.

When it happens

Trigger: Running a pipeline with --runner=PortableRunner and no --job_endpoint, so create_job_service -> default_job_server hits the base-class NotImplementedError.

Common situations: Testing against a local Beam job service but forgetting the endpoint flag; switching a pipeline from DirectRunner to PortableRunner without adding job-service configuration; tutorials that assume FlinkRunner's default endpoint.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/portability/portable_runner.py:268

class PortableRunner(runner.PipelineRunner):
  """
    Experimental: No backward compatibility guaranteed.
    A BeamRunner that executes Python pipelines via the Beam Job API.

    This runner is a stub and does not run the actual job.
    This runner schedules the job on a job service. The responsibility of
    running and managing the job lies with the job service used.
  """
  def __init__(self):
    self._dockerized_job_server: Optional[job_server.JobServer] = None

  @staticmethod
  def _create_environment(options: PipelineOptions) -> environments.Environment:
    return environments.Environment.from_options(
        options.view_as(PortableOptions))

  def default_job_server(self, options):
    raise NotImplementedError(
        'You must specify a --job_endpoint when using --runner=PortableRunner. '
        'Alternatively, you may specify which portable runner you intend to '
        'use, such as --runner=FlinkRunner or --runner=SparkRunner.')

  def create_job_service_handle(self, job_service, options) -> JobServiceHandle:
    return JobServiceHandle(job_service, options)

  def create_job_service(self, options: PipelineOptions) -> JobServiceHandle:
    """
    Start the job service and return a `JobServiceHandle`
    """
    job_endpoint = options.view_as(PortableOptions).job_endpoint
    if job_endpoint:
      if job_endpoint == 'embed':
        server: job_server.JobServer = job_server.EmbeddedJobServer()
      else:
        job_server_timeout = options.view_as(PortableOptions).job_server_timeout
        server = job_server.ExternalJobServer(job_endpoint, job_server_timeout)

View on GitHub (pinned to 12126d8942)