apache/beam · error · TypeError

FlattenWith only takes other PCollections and PTransforms…

Error message

FlattenWith only takes other PCollections and PTransforms, got {other}

What it means

Raised by FlattenWith.expand when one of the stored _others is neither a PCollection nor a PTransform. FlattenWith only knows how to append a PCollection directly or apply a PTransform to the source pipeline's input; any other object cannot contribute a collection to the flatten.

Solutions

1. Pass PCollections directly, unpacking lists: beam.FlattenWith(*other_pcolls).
2. If merging with a transform's output, pass the PTransform: | beam.FlattenWith(beam.Create([...])).
3. Ensure each argument is a PCollection (e.g. PValue vs PCollection distinction) or a PTransform.
4. Unpack any container variables before passing.

Example fix

// before
merged = pc | beam.FlattenWith(other_pcolls)  # other_pcolls is a list
// after
merged = pc | beam.FlattenWith(*other_pcolls)

assert all(isinstance(o, (PCollection, PTransform)) for o in others), 'FlattenWith accepts only PCollections and PTransforms'

def is_flatten_with_arg(o):
    return isinstance(o, (PCollection, PTransform))

try:
    merged = pc | beam.FlattenWith(*others)
except TypeError as e:
    log.error('Bad FlattenWith argument: %s', e)

When it happens

Trigger: pc | beam.FlattenWith(some_raw_list), FlattenWith(42), or accidentally passing a list of PCollections instead of each PCollection individually; passing a PValue that is not a PCollection.

Common situations: Collecting things to merge into a plain list and passing the list itself; mixing string/table references (other runners/SQL) with PCollection arguments; typos where a variable holds the wrong object.

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

• A sink must inherit iobase.Sink, iobase.NativeSink, or be a…
• Cannot encode payload for WriteToPubSub. Expected valid…
• Cannot interpret as seconds.
• Cannot interpret as subseconds.
• compression_type must be CompressionType object but was

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

Appendix: source

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

  This is equivalent to creating a tuple containing both the input and the
  other PCollection(s), but has the advantage that it can be more easily used
  inline.

  Root PTransforms can be passed as well as PCollections, in which case their
  outputs will be flattened.
  """
  def __init__(self, *others):
    self._others = others

  def expand(self, pcoll):
    pcolls = [pcoll]
    for other in self._others:
      if isinstance(other, pvalue.PCollection):
        pcolls.append(other)
      elif isinstance(other, PTransform):
        pcolls.append(pcoll.pipeline | other)
      else:
        raise TypeError(
            'FlattenWith only takes other PCollections and PTransforms, '
            f'got {other}')
    return tuple(pcolls) | Flatten()


class Create(PTransform):
  """A transform that creates a PCollection from an iterable."""
  def __init__(self, values, reshuffle=True):
    """Initializes a Create transform.

    Args:
      values: An object of values for the PCollection
    """
    super().__init__()
    if isinstance(values, (str, bytes)):
      raise TypeError(
          'PTransform Create: Refusing to treat string as '
          'an iterable. (string=%r)' % values)

View on GitHub (pinned to 12126d8942)