apache/beam · error · ValueError

bucket_boundaries requires length_fn to be set.

Error message

bucket_boundaries requires length_fn to be set.

What it means

Raised when BatchElements is given bucket_boundaries (fixed size buckets) without a length_fn. Bucket-based size estimation needs to know how to measure each element's size; without length_fn the estimator cannot assign elements to buckets, so construction fails.

Solutions

Provide length_fn, e.g. BatchElements(bucket_boundaries=[10, 100], length_fn=lambda x: len(x)).
If per-element size is not meaningful for your data, drop bucket_boundaries and use target_batch_overhead/target_batch_duration_secs instead.
Centralize the pairing of bucket_boundaries and length_fn in a helper so they are always passed together.

Example fix

// before
beam.BatchElements(bucket_boundaries=[10, 100, 1000])

// after
beam.BatchElements(bucket_boundaries=[10, 100, 1000],
                   length_fn=lambda batch: len(batch))

Defensive patterns

Strategy: validation

Validate before calling

if bucket_boundaries is not None and length_fn is None:
    raise ValueError('bucket_boundaries requires length_fn')

Type guard

def valid_bucket_config(boundaries, length_fn):
    return boundaries is None or (callable(length_fn) and boundaries is not None)

Try / catch

try:
    t = beam.BatchElements(bucket_boundaries=bounds)
except ValueError as e:
    if 'length_fn' in str(e):
        t = beam.BatchElements(bucket_boundaries=bounds, length_fn=lambda x: len(x))
    else:
        raise

Prevention

Always pass bucket_boundaries and length_fn together via a shared helper/factory.
Use buckets only when per-element size (bytes/records) is well defined; otherwise use target_* parameters.
Add a construction-time unit test for every BatchElements configuration.

When it happens

Trigger: BatchElements(bucket_boundaries=[10, 100, 1000]) without length_fn.

Common situations: Copying a bucket example and dropping the length_fn; assuming elements are counted by default when a custom notion of size (bytes, records, weight) is required.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

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

Appendix: source

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

  _DEFAULT_BUCKET_BOUNDARIES = [16, 32, 64, 128, 256, 512]

  def __init__(
      self,
      min_batch_size=1,
      max_batch_size=10000,
      target_batch_overhead=.05,
      target_batch_duration_secs=10,
      target_batch_duration_secs_including_fixed_cost=None,
      max_batch_duration_secs=None,
      *,
      element_size_fn=lambda x: 1,
      variance=0.25,
      clock=time.time,
      record_metrics=True,
      length_fn=None,
      bucket_boundaries=None):
    if bucket_boundaries is not None and length_fn is None:
      raise ValueError('bucket_boundaries requires length_fn to be set.')
    if bucket_boundaries is not None:
      if (not bucket_boundaries or any(b <= 0 for b in bucket_boundaries) or
          bucket_boundaries != sorted(bucket_boundaries)):
        raise ValueError(
            'bucket_boundaries must be a non-empty sorted list of '
            'positive values.')
    self._batch_size_estimator = _BatchSizeEstimator(
        min_batch_size=min_batch_size,
        max_batch_size=max_batch_size,
        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

View on GitHub (pinned to 12126d8942)