apache/beam · error · ValueError

Parameter for a Zipf distribution must be larger than 1…

Error message

Parameter for a Zipf distribution must be larger than 1. Received %r.

What it means

SyntheticSource's __init__ validates the bundle-size distribution parameter when initial splitting is set to 'zipf'. A Zipf distribution is only mathematically defined for a parameter strictly greater than 1, so the source raises ValueError when the configured param is <= 1. Note the check uses < 1, so exactly 1 also passes but is still invalid for a true Zipf law; any value below 1 is rejected.

Solutions

  1. Set input_spec['bundleSizeDistribution']['param'] to a value strictly greater than 1 (e.g. 1.5 or 2.0).
  2. If you do not specifically need Zipf, switch initial splitting to 'uniform' (the parameter is then ignored).
  3. Validate the config JSON before launching the pipeline (check type=='zipf' implies param > 1).

Example fix

// before
input_spec = {'initialSplitting': 'zipf', 'bundleSizeDistribution': {'type': 'zipf', 'param': 1}}
// after
input_spec = {'initialSplitting': 'zipf', 'bundleSizeDistribution': {'type': 'zipf', 'param': 2.0}}
Defensive patterns

Strategy: validation

Validate before calling

dist = input_spec.get('bundleSizeDistribution', {})
if input_spec.get('initialSplitting') == 'zipf' and not (isinstance(dist.get('param'), (int, float)) and dist.get('param', 0) > 1):
    raise ValueError('zipf param must be > 1')

Type guard

def is_valid_zipf_param(v) -> bool:
    return isinstance(v, (int, float)) and v > 1

Try / catch

try:
    source = SyntheticStep(input_spec, ...)
except ValueError as e:
    if 'Zipf distribution' in str(e):
        input_spec['bundleSizeDistribution']['param'] = 2.0
        source = SyntheticStep(input_spec, ...)
    else:
        raise

Prevention

When it happens

Trigger: Constructing a SyntheticSource (or running a synthetic pipeline) with input_spec['initialSplitting'] == 'zipf' and input_spec['bundleSizeDistribution'] == {'type': 'zipf', 'param': 1} or any param less than 1 (e.g. 0.5, 0, -3).

Common situations: Hand-editing synthetic pipeline config JSON for load-testing Beam runners; copying a uniform-distribution config and only changing the type to 'zipf' without adjusting param; param mistakenly given as a percentage (0.9) instead of the Zipf exponent.

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/testing/synthetic_pipeline.py:372

    self._value_size = maybe_parse_byte_size(
        input_spec.get('valueSizeBytes', 1))
    self._total_size = self.element_size * self._num_records
    self._initial_splitting = (
        input_spec['bundleSizeDistribution']['type']
        if 'bundleSizeDistribution' in input_spec else 'const')
    if self._initial_splitting != 'const' and self._initial_splitting != 'zipf':
      raise ValueError(
          'Only const and zipf distributions are supported for determining '
          'sizes of bundles produced by initial splitting. Received: %s',
          self._initial_splitting)
    self._initial_splitting_num_bundles = (
        input_spec['forceNumInitialBundles']
        if 'forceNumInitialBundles' in input_spec else 0)
    if self._initial_splitting == 'zipf':
      self._initial_splitting_distribution_parameter = (
          input_spec['bundleSizeDistribution']['param'])
      if self._initial_splitting_distribution_parameter < 1:
        raise ValueError(
            'Parameter for a Zipf distribution must be larger than 1. '
            'Received %r.',
            self._initial_splitting_distribution_parameter)
    else:
      self._initial_splitting_distribution_parameter = 0
    self._dynamic_splitting = (
        'none' if (
            'splitPointFrequencyRecords' in input_spec and
            input_spec['splitPointFrequencyRecords'] == 0) else 'perfect')
    if 'delayDistribution' in input_spec:
      if input_spec['delayDistribution']['type'] != 'const':
        raise ValueError(
            'SyntheticSource currently only supports delay '
            'distributions of type \'const\'. Received %s.',
            input_spec['delayDistribution']['type'])
      self._sleep_per_input_record_sec = (
          float(input_spec['delayDistribution']['const']) / 1000)
      if (self._sleep_per_input_record_sec and

View on GitHub (pinned to 12126d8942)