apache/beam · error · ValueError

Unable to run pipeline with requirement

Error message

Unable to run pipeline with requirement: %s

What it means

PipelineRunner.check_requirements() validates that every requirement URN in the pipeline proto is supported by the runner's supported_requirements set. If a pipeline declares a requirement (e.g. from a side input, stateful DoFn, or test stream) the target runner cannot handle, this ValueError is raised before execution.

Solutions

  1. Remove the pipeline feature that adds the unsupported requirement
  2. Use a runner that supports the requirement (check supported_requirements for your runner)
  3. Upgrade apache_beam / the runner, as newer versions implement more requirements

Example fix

// before
# pipeline uses stateful DoFn on a runner without STATEFUL_PROCESSING support
result = pipeline.run()
// after
# switch to a runner supporting the requirement, or refactor the DoFn to be stateless
pipeline = beam.Pipeline(runner='DirectRunner')
Defensive patterns

Strategy: try-catch

Validate before calling

unsupported = set(pipeline_proto.requirements) - set(runner.supported_requirements)
if unsupported:
    print('runner lacks:', unsupported)

Try / catch

try:
    runner.check_requirements(pipeline_proto, runner.supported_requirements)
except ValueError as e:
    # inspect e for the unsupported requirement URN, pick another runner
    raise

Prevention

When it happens

Trigger: Calling run_portable_pipeline (via check_requirements) with a pipeline proto whose pipeline_proto.requirements contains a URN not present in supported_requirements.

Common situations: Switching to a runner that lacks support for a feature the pipeline uses (e.g. runners that don't implement a specific requirement); building pipelines with state/timers/splittable DoFns and running them on a runner without those capabilities.

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

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/runner.py:213

    return transform.expand(input)

  def is_fnapi_compatible(self):
    """Whether to enable the beam_fn_api experiment by default."""
    return True

  def check_requirements(
      self,
      pipeline_proto: beam_runner_api_pb2.Pipeline,
      supported_requirements: Iterable[str]):
    """Check that this runner can satisfy all pipeline requirements."""

    # Imported here to avoid circular dependencies.
    # pylint: disable=wrong-import-order, wrong-import-position
    from apache_beam.runners.portability.fn_api_runner import translations
    supported_requirements = set(supported_requirements)
    for requirement in pipeline_proto.requirements:
      if requirement not in supported_requirements:
        raise ValueError(
            'Unable to run pipeline with requirement: %s' % requirement)
    for transform in pipeline_proto.components.transforms.values():
      if transform.spec.urn == common_urns.primitives.TEST_STREAM.urn:
        if common_urns.primitives.TEST_STREAM.urn not in supported_requirements:
          raise NotImplementedError(transform.spec.urn)
      elif transform.spec.urn in translations.PAR_DO_URNS:
        payload = beam_runner_api_pb2.ParDoPayload.FromString(
            transform.spec.payload)
        for timer in payload.timer_family_specs.values():
          if timer.time_domain not in (
              beam_runner_api_pb2.TimeDomain.EVENT_TIME,
              beam_runner_api_pb2.TimeDomain.PROCESSING_TIME):
            raise NotImplementedError(timer.time_domain)

  def default_pickle_library_override(self):
    """Default pickle library, can be overridden by runner implementation."""
    return None

View on GitHub (pinned to 12126d8942)