apache/beam · error · RuntimeError

All dicts in batch must have the same keys. extra keys

Error message

All dicts in batch must have the same keys. extra keys: {extra_keys}, missing keys: {missing_keys}

What it means

Within a batch, every dict must have the identical set of keys; the first element defines the expected keys. If a later element has extra or missing keys, this RuntimeError reports the differing key sets.

Solutions

1. Normalize all records to the same schema before batching (add None/default values for missing keys, drop extras).
2. Use a beam.Map to project each record onto a fixed key set matching the transform's columns.
3. Fix the upstream producer so it always emits the same fields.

Example fix

// before
beam.Create([{'text': 'a'}, {'text': 'b', 'meta': 'm'}])
// after
beam.Create([{'text': 'a', 'meta': None}, {'text': 'b', 'meta': 'm'}])

keys = {frozenset(d.keys()) for d in batch}
assert len(keys) == 1, f'Heterogeneous batch keys: {keys}'

def has_uniform_keys(batch) -> bool:
    if not batch:
        return True
    expected = set(batch[0].keys())
    return all(set(d.keys()) == expected for d in batch if isinstance(d, dict))

try:
    out = data | MLTransform(...)
except RuntimeError as e:
    if 'same keys' in str(e):
        raise ValueError(f'Upstream schema drift detected: {e}') from e
    raise

When it happens

Trigger: Batches where some records contain optional fields others lack (e.g. {'text': ...} vs {'text': ..., 'metadata': ...}), or rows built dynamically from heterogeneous sources before MLTransform.

Common situations: Joining documents from multiple sources into one PCollection; optional metadata fields present in only some records; schema drift between pipeline stages.

Understand the failure class

Background: Schema validation failed / invalid input schema: payload rejected because its shape doesn't match the expected schema — this error's family across 28 libraries.

Related errors

• A schema is required to write non-schema'd data.
• An explicit schema is required to write non-schema'd…
• Arrow map key field cannot be nullable
• Attempted to encode null for non-nullable field
• Attribute fields not found in schema fields

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

Appendix: source

Thrown at sdks/python/apache_beam/ml/transforms/base.py:204

                                                  beam.Row]]) -> list[str]:
  """Extract text from specified columns in batch."""
  if batch and hasattr(batch[0], '_asdict'):
    batch = [row._asdict() if hasattr(row, '_asdict') else row for row in batch]

  if not batch or not isinstance(batch[0], dict):
    raise TypeError(
        'Expected data to be dicts, got '
        f'{type(batch[0])} instead.')
  result = []
  expected_keys = set(batch[0].keys())
  expected_columns = set(columns)
  # Process one batch item at a time
  for item in batch:
    item_keys = item.keys() if isinstance(item, dict) else set()
    if set(item_keys) != expected_keys:
      extra_keys = item_keys - expected_keys
      missing_keys = expected_keys - item_keys
      raise RuntimeError(
          f'All dicts in batch must have the same keys. '
          f'extra keys: {extra_keys}, '
          f'missing keys: {missing_keys}')
    missing_columns = expected_columns - item_keys
    if (missing_columns):
      raise RuntimeError(
          f'Data does not contain the following columns '
          f': {missing_columns}.')

    # Get all columns for this item
    for col in columns:
      if isinstance(item, dict):
        result.append(item[col])
  return result


def _dict_output_fn(
    columns: Sequence[str],

View on GitHub (pinned to 12126d8942)