apache/beam · error · NotImplementedError

Unable to extract PValue inputs from

Error message

Unable to extract PValue inputs from %s; either %s does not accept inputs of this format, or it does not properly override _extract_input_pvalues

What it means

Beam asks the transform to decompose its input via _extract_input_pvalues. If that call raises TypeError (the transform can't iterate/consume the input format), _apply_internal raises NotImplementedError saying it cannot extract PValue inputs from the given value — either the input format is unsupported or the custom transform didn't properly override _extract_input_pvalues.

Solutions

Pass a PCollection (or PTransform-accepted PValue structure) as the input.
If your transform accepts tuples/dicts of PCollections, override _extract_input_pvalues to return (pvalueish, dict-of-leaf-pvalues).
Unwrap raw lists/dicts into PCollections with beam.Create before applying.
Check which part of the input is not a PValue; the message names the pvalueish and transform.

Example fix

// before
class MyT(PTransform):
  def expand(self, pcolls):  # pcolls is a tuple, no _extract_input_pvalues override
    ...
p.apply(MyT(), (pc1, pc2))
// after
class MyT(PTransform):
  def _extract_input_pvalues(self, pvalueish):
    return pvalueish, {'a': pvalueish[0], 'b': pvalueish[1]}
  def expand(self, pcolls):
    ...
p.apply(MyT(), (pc1, pc2))

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam.pvalue import PValue
def ensure_pvalue_inputs(x):
    if isinstance(x, PValue):
        return x
    if isinstance(x, (tuple, list)):
        assert all(isinstance(i, PValue) for i in x), 'non-PValue member in input'
    else:
        raise TypeError('input must be a PValue or tuple/list of PValues')
    return x

Try / catch

try:
    out = pipeline.apply(t, pvalueish)
except NotImplementedError as e:
    if 'extract PValue inputs' in str(e):
        raise ValueError(f'{t} cannot consume {pvalueish!r}; wrap raw data with beam.Create') from e

Prevention

Feed PCollections, not raw collections, into transforms
Override _extract_input_pvalues for composites with tuple/dict inputs
Use beam.Create for in-memory data

When it happens

Trigger: Passing a non-PValue container (list of values, dict of raw data, plain iterable) to pipeline.apply for a transform that doesn't accept/extract it; a custom PTransform lacking a correct _extract_input_pvalues override for unusual inputs (e.g. tuple inputs).

Common situations: Passing raw Python collections instead of PCollections into composite transforms; custom PTransforms handling tuple/dict inputs without overriding _extract_input_pvalues; feeding side-input-style values as main inputs.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/pipeline.py:779

            '"auto_unique_labels" to automatically generate unique '
            'transform labels. Note "auto_unique_labels" '
            'could cause data loss when updating a pipeline or '
            'reloading the job state. This is not recommended for '
            'streaming jobs.' % full_label)
    self.applied_labels.add(full_label)

    if pvalueish is None:
      full_label = self._current_transform().full_label
      raise TypeCheckError(
          f'Transform "{full_label}" was applied to the output of '
          f'an object of type None.')

    pvalueish, inputs = transform._extract_input_pvalues(pvalueish)
    try:
      if not isinstance(inputs, dict):
        inputs = {str(ix): input for (ix, input) in enumerate(inputs)}
    except TypeError:
      raise NotImplementedError(
          'Unable to extract PValue inputs from %s; either %s does not accept '
          'inputs of this format, or it does not properly override '
          '_extract_input_pvalues' % (pvalueish, transform))
    for t, leaf_input in inputs.items():
      if not isinstance(leaf_input, pvalue.PValue) or not isinstance(t, str):
        raise NotImplementedError(
            '%s does not properly override _extract_input_pvalues, '
            'returned %s from %s' % (transform, inputs, pvalueish))

    current = AppliedPTransform(
        self._current_transform(),
        transform,
        full_label,
        inputs,
        None,
        annotations=self._current_annotations())
    self._current_transform().add_part(current)

View on GitHub (pinned to 12126d8942)