apache/beam · error · NotImplementedError

Requires stateful processing (BEAM-2687)

Error message

Requires stateful processing (BEAM-2687)

What it means

BatchReassignWindows (an internal batching transform in apache_beam.transforms.util) rejects pipelines running on a streaming runner, because grouping into batches requires stateful processing support (tracked under JIRA BEAM-2687). The check happens in expand() when the runner reports is_streaming. The library throws NotImplementedError because the operation simply cannot execute on the given runner.

Solutions

1. Run the pipeline with a batch runner (e.g. DirectRunner, DataflowRunner batch mode) for this stage
2. Use a streaming-native batching/windowing alternative (e.g. GroupIntoBatches with streaming support or fixed windows + GBK)
3. Check the pipeline options: remove is_streaming settings or move the transform into a separate batch job

Example fix

// before
with beam.Pipeline(options=streaming_options) as p:
  _ = (p | beam.Create(data) | util.BatchReassignWindows())
// after
# Use GroupIntoBatches, which supports streaming
_ = (p | beam.KVToTuple() | util.GroupIntoBatches(batch_size=100))

if getattr(pipeline.runner, 'is_streaming', False):
    raise RuntimeError('BatchReassignWindows requires a batch runner')

When it happens

Trigger: Applying this batching transform to a PCollection in a pipeline whose runner has is_streaming=True (e.g. Flink/Spark/Beam StreamingDataflow without stateful processing), so the transform's expand() immediately raises before any data flows.

Common situations: Developers take a batch-oriented pipeline that works with DirectRunner and switch runner to a streaming job; or they build a pipeline parameterized by runner and accidentally route the batching stage into a streaming context.

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

• Assigning an index is not yet supported. Consider using…
• by
• collecting metrics will come later!
• concat(ignore_index)
• concat(levels)

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/util.py:1115

        target_batch_overhead=target_batch_overhead,
        target_batch_duration_secs=target_batch_duration_secs,
        target_batch_duration_secs_including_fixed_cost=(
            target_batch_duration_secs_including_fixed_cost),
        variance=variance,
        clock=clock,
        record_metrics=record_metrics)
    self._element_size_fn = element_size_fn
    self._max_batch_dur = max_batch_duration_secs
    self._clock = clock
    self._length_fn = length_fn
    if length_fn is not None and bucket_boundaries is None:
      self._bucket_boundaries = self._DEFAULT_BUCKET_BOUNDARIES
    else:
      self._bucket_boundaries = bucket_boundaries

  def expand(self, pcoll):
    if getattr(pcoll.pipeline.runner, 'is_streaming', False):
      raise NotImplementedError("Requires stateful processing (BEAM-2687)")
    elif self._max_batch_dur is not None:
      coder = coders.registry.get_coder(pcoll)
      if self._length_fn is not None:
        keying_dofn = WithLengthBucketKey(
            self._length_fn, self._bucket_boundaries)
      else:
        keying_dofn = WithSharedKey()
      return pcoll | ParDo(keying_dofn) | ParDo(
          _pardo_stateful_batch_elements(
              coder,
              self._batch_size_estimator,
              self._max_batch_dur,
              self._clock))
    elif pcoll.windowing.is_default():
      # This is the same logic as _GlobalWindowsBatchingDoFn, but optimized
      # for that simpler case.
      return pcoll | ParDo(
          _GlobalWindowsBatchingDoFn(

View on GitHub (pinned to 12126d8942)