apache/beam · error · TypeError

Inputs to Flatten cannot include an iterable of…

Error message

Inputs to Flatten cannot include an iterable of PCollections. (input at index {idx}: "{item}")

What it means

Raised by Flatten._extract_input_pvalues when one element of the input is itself a list/tuple containing PCollections — i.e. a nested iterable of PCollections inside the flatten input. Beam treats this as an always-a-user-error structure and rejects it with a TypeError, reporting the index and the offending item.

Solutions

  1. Flatten the nested structure before piping: inputs = [pc1] + [pc2, pc3].
  2. Use itertools.chain or a comprehension to produce a flat list of PCollections.
  3. Apply Flatten twice if nested grouping is intentional: flatten inner groups first, then the outer list.
  4. Assert all elements are PCollection instances before calling Flatten.

Example fix

// before
merged = [pc1, [pc2, pc3]] | beam.Flatten()
// after
merged = [pc1, pc2, pc3] | beam.Flatten()
Defensive patterns

Strategy: validation

Validate before calling

for i, item in enumerate(inputs):
    assert not (isinstance(item, (list, tuple)) and any(isinstance(s, PCollection) for s in item)), f'nested PCollections at index {i}'

Type guard

def is_flat_pcoll_list(v):
    return not any(isinstance(x, (list, tuple)) for x in v)

Try / catch

try:
    merged = inputs | beam.Flatten()
except TypeError as e:
    log.error('Nested PCollection iterable: %s', e)

Prevention

When it happens

Trigger: Passing something like [pc1, [pc2, pc3]] or (pc1, (pc2, pc3)) to Flatten; grouping PCollections into sub-lists and flattening the outer list; results of functions that return lists of PCollections inserted into a parent list.

Common situations: Building inputs programmatically where some entries are already collections of PCollections; mixing grouped and ungrouped inputs from helper functions.

Understand the failure class

Background: "Invalid ... format", "must be in format X", "does not look like a ..." — invalid argument format errors across CLI tools and libraries — this error's family across 17 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/core.py:4119

    self.pipeline = kwargs.pop(
        'pipeline', None)  # type: typing.Optional[Pipeline]
    if kwargs:
      raise ValueError('Unexpected keyword arguments: %s' % list(kwargs))

  def _extract_input_pvalues(self, pvalueish):
    try:
      pvalueish = tuple(pvalueish)
    except TypeError:
      raise ValueError(
          'Input to Flatten must be an iterable. '
          'Got a value of type %s instead.' % type(pvalueish))

    # Spot check to see if any of the items are iterables of PCollections
    # and raise an error if so. This is always a user-error
    for idx, item in enumerate(pvalueish):
      if isinstance(item, (list, tuple)) and any(
          isinstance(sub_item, pvalue.PCollection) for sub_item in item):
        raise TypeError(
            'Inputs to Flatten cannot include an iterable of PCollections. '
            f'(input at index {idx}: "{item}")')
    return pvalueish, pvalueish

  def expand(self, pcolls):
    windowing = self.get_windowing(pcolls)
    for pcoll in pcolls:
      self._check_pcollection(pcoll)
      if pcoll.windowing != windowing:
        _LOGGER.warning(
            'All input pcollections must have the same window. Windowing for '
            'flatten set to %s, windowing of pcoll %s set to %s',
            windowing,
            pcoll,
            pcoll.windowing)
    is_bounded = all(pcoll.is_bounded for pcoll in pcolls)
    return pvalue.PCollection(self.pipeline, is_bounded=is_bounded)

View on GitHub (pinned to 12126d8942)