apache/beam · error · ValueError

Please specify either `dask_npartitions` or…

Error message

Please specify either `dask_npartitions` or `dask_parition_size` but not both: npartitions=%r, partition_size=%r.

What it means

When evaluating a beam.Create transform, the Dask evaluator requires exactly one of `dask_npartitions` or `dask_partition_size` to control bag chunking. Passing both is ambiguous, so it raises ValueError.

Solutions

Remove one of the two options from your PipelineOptions/flags.
Keep only `dask_npartitions` if you know the desired partition count.
Keep only `dask_partition_size` to let the evaluator compute partitions from total size.

Example fix

// before
options = PipelineOptions(['--dask_npartitions=8', '--dask_partition_size=50MB'])
// after
options = PipelineOptions(['--dask_npartitions=8'])

Defensive patterns

Strategy: validation

Validate before calling

opts = dask_options.get_all_options(drop_default=True, current_only=True)
if opts.get('dask_npartitions') and opts.get('dask_partition_size'):
    raise ValueError('Set only one of dask_npartitions/dask_partition_size')

Try / catch

try:
    pipeline.run()
except ValueError as e:
    if 'but not both' in str(e):
        logging.error('Remove one of dask_npartitions/dask_partition_size')
    raise

Prevention

Set exactly one partitioning knob in your pipeline options.
Centralize Dask option construction in one config module.
Grep deployment flags for both keys before launching.

When it happens

Trigger: Setting both `dask_npartitions` and `dask_partition_size` options (e.g. via PipelineOptions --dask_npartitions and --dask_partition_size) for a pipeline containing beam.Create.

Common situations: Config mistakes where users copy an example and add both tuning knobs, or leftover flags from earlier experiments.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/dask/transform_evaluator.py:165


class NoOp(DaskBagOp):
  """An identity on a dask bag: returns the input as-is."""
  def apply(self, input_bag: OpInput, side_inputs: OpSide = None) -> db.Bag:
    return input_bag


class Create(DaskBagOp):
  """The beginning of a Beam pipeline; the input must be `None`."""
  def apply(self, input_bag: OpInput, side_inputs: OpSide = None) -> db.Bag:
    assert input_bag is None, 'Create expects no input!'
    original_transform = t.cast(_Create, self.transform)
    items = original_transform.values

    npartitions = self.bag_kwargs.get('npartitions')
    partition_size = self.bag_kwargs.get('partition_size')
    if npartitions and partition_size:
      raise ValueError(
          f'Please specify either `dask_npartitions` or '
          f'`dask_parition_size` but not both: '
          f'{npartitions=}, {partition_size=}.')
    if not npartitions and not partition_size:
      # partition_size is inversely related to `npartitions`.
      # Ideal "chunk sizes" in dask are around 10-100 MBs.
      # Let's hope ~128 items per partition is around this
      # memory overhead.
      default_size = 128
      partition_size = max(default_size, math.ceil(math.sqrt(len(items)) / 10))
      if partition_size == default_size:
        _LOGGER.warning(
            'The new default partition size is %d, it used to be 1 '
            'in previous DaskRunner versions.' % default_size)

    return db.from_sequence(
        items, npartitions=npartitions, partition_size=partition_size)

View on GitHub (pinned to 12126d8942)