apache/beam · error · RuntimeError

Runtime exception

Error message

Runtime exception: %s

What it means

ApproximateUnique's AddInput transform wraps any exception raised while hashing/encoding an element into a RuntimeError with the original message embedded. This is a coarse wrapper added so worker-side failures surface with context, but it obscures the underlying error (e.g. a coder or hash function failure). The real cause is the wrapped exception text after 'Runtime exception:'.

Solutions

  1. Read the text after 'Runtime exception: ' to find the real error; fix that underlying cause first.
  2. Verify the element type flowing into ApproximateUnique matches its coder argument; pass an explicit coder if needed.
  3. If using a non-default hash, confirm the required hash library is installed in the runner environment or revert to the default hash.
  4. Test the hash of one element locally: hasher(coder.encode(sample)) to reproduce before launching the pipeline.

Example fix

// before
pcoll | ApproximateUnique.Globally(1000)  # elements are dicts, default coder fails
// after
pcoll | Map(lambda d: json.dumps(d, sort_keys=True)) | ApproximateUnique.Globally(1000)
Defensive patterns

Strategy: try-catch

Validate before calling

from apache_beam import coders
try:
    encoded = my_coder.encode(sample_element)
except Exception as e:
    raise ValueError(f'Element not encodable by {my_coder}: {e}')

Type guard

def encodable(coder, element):
    try:
        coder.encode(element)
        return True
    except Exception:
        return False

Try / catch

try:
    result = pcoll | ApproximateUnique.Globally(n)
except RuntimeError as e:
    logger.error('ApproximateUnique underlying failure: %s', e)
    # inspect inner message: coder vs hash_fn cause

Prevention

When it happens

Trigger: Calling apache_beam.transforms.stats.ApproximateUnique combine_fn's add_input (directly or via a pipeline) when self._coder.encode(element) raises (element not encodable by the configured coder) or self._hash_fn raises (custom hash function fails on the encoded bytes).

Common situations: Pipelines where the element type does not match the coder supplied to ApproximateUnique.Globally()/PerKey(); custom --hash_fn values (e.g. 'cityhash'/'farmhash' selected but the C extension is unavailable in the environment); elements containing types the default varint coder cannot handle.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/stats.py:263

  """
  def __init__(self, sample_size, coder):
    self._sample_size = sample_size
    coder = coders.typecoders.registry.verify_deterministic(
        coder, 'ApproximateUniqueCombineFn')

    self._coder = coder
    self._hash_fn = _get_default_hash_fn()

  def create_accumulator(self, *args, **kwargs):
    return _LargestUnique(self._sample_size)

  def add_input(self, accumulator, element, *args, **kwargs):
    try:
      hashed_value = self._hash_fn(self._coder.encode(element))
      accumulator.add(hashed_value)
      return accumulator
    except Exception as e:
      raise RuntimeError("Runtime exception: %s" % e)

  # created an issue https://github.com/apache/beam/issues/19459 to speed up
  # merge process.
  def merge_accumulators(self, accumulators, *args, **kwargs):
    merged_accumulator = self.create_accumulator()
    for accumulator in accumulators:
      for i in accumulator._sample_heap:
        merged_accumulator.add(i)

    return merged_accumulator

  @staticmethod
  def extract_output(accumulator):
    return accumulator.get_estimate()

  def display_data(self):
    return {'sample_size': self._sample_size}

View on GitHub (pinned to 12126d8942)