apache/beam · error · ValueError

Unexpected keyword arguments

Error message

Unexpected keyword arguments: %s

What it means

Raised by Flatten's __init__ when any keyword argument other than 'pipeline' is passed. Flatten accepts no user-facing configuration besides the optional pipeline argument; unknown kwargs indicate a mistake (e.g. misremembered option names or args meant for another transform).

Solutions

  1. Remove unexpected keyword arguments; Flatten takes PCollections via the pipe operator or a list, not options.
  2. Merge PCollections as: flattened = pcoll1 | pcoll2 | pcoll3 or pc | beam.Flatten(pcoll_list).
  3. Check the API docs for the installed Beam version — the kwargs you pass may belong to a different transform.

Example fix

// before
result = (pc1, pc2) | beam.Flatten(pipeline=p)
// after
result = (pc1, pc2) | beam.Flatten()
Defensive patterns

Strategy: type-guard

Validate before calling

kwargs = {'pipeline': p, 'bogus': 1}
unknown = set(kwargs) - {'pipeline'}
assert not unknown, f'unknown Flatten kwargs: {unknown}'

Type guard

def is_flatten_kwargs(kw):
    return all(k == 'pipeline' for k in kw)

Try / catch

try:
    merged = (pc1, pc2) | beam.Flatten(**kw)
except ValueError as e:
    log.error('Flatten kwargs rejected: %s', e)

Prevention

When it happens

Trigger: beam.Flatten(pcolls=[...]), Flatten(parallel_input=True) or any other legacy/imagined kwargs; dynamically forwarding **options into Flatten.

Common situations: Porting Spark's union options or old Beam args onto Flatten; passing **kwargs captured from a wrapper function; confusing Flatten with CombineGlobally options.

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/856d58e23638fa27. Report an issue: GitHub.

Appendix: source

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

  """Merges several PCollections into a single PCollection.

  Copies all elements in 0 or more PCollections into a single output
  PCollection. If there are no input PCollections, the resulting PCollection
  will be empty (but see also kwargs below).

  Args:
    **kwargs: Accepts a single named argument "pipeline", which specifies the
      pipeline that "owns" this PTransform. Ordinarily Flatten can obtain this
      information from one of the input PCollections, but if there are none (or
      if there's a chance there may be none), this argument is the only way to
      provide pipeline information and should be considered mandatory.
  """
  def __init__(self, **kwargs):
    super().__init__()
    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

View on GitHub (pinned to 12126d8942)