{"record":{"id":"2565e4b82ac0c229","repo":"apache/beam","slug":"all-dicts-in-batch-must-have-the-same-keys-extra-keys-extra","errorCode":null,"errorMessage":"All dicts in batch must have the same keys. extra keys: {extra_keys}, missing keys: {missing_keys}","messagePattern":"All dicts in batch must have the same keys\\. extra keys: (.+?), missing keys: (.+?)","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/ml/transforms/base.py","lineNumber":204,"sourceCode":"                                                  beam.Row]]) -> list[str]:\n  \"\"\"Extract text from specified columns in batch.\"\"\"\n  if batch and hasattr(batch[0], '_asdict'):\n    batch = [row._asdict() if hasattr(row, '_asdict') else row for row in batch]\n\n  if not batch or not isinstance(batch[0], dict):\n    raise TypeError(\n        'Expected data to be dicts, got '\n        f'{type(batch[0])} instead.')\n  result = []\n  expected_keys = set(batch[0].keys())\n  expected_columns = set(columns)\n  # Process one batch item at a time\n  for item in batch:\n    item_keys = item.keys() if isinstance(item, dict) else set()\n    if set(item_keys) != expected_keys:\n      extra_keys = item_keys - expected_keys\n      missing_keys = expected_keys - item_keys\n      raise RuntimeError(\n          f'All dicts in batch must have the same keys. '\n          f'extra keys: {extra_keys}, '\n          f'missing keys: {missing_keys}')\n    missing_columns = expected_columns - item_keys\n    if (missing_columns):\n      raise RuntimeError(\n          f'Data does not contain the following columns '\n          f': {missing_columns}.')\n\n    # Get all columns for this item\n    for col in columns:\n      if isinstance(item, dict):\n        result.append(item[col])\n  return result\n\n\ndef _dict_output_fn(\n    columns: Sequence[str],","sourceCodeStart":186,"sourceCodeEnd":222,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/ml/transforms/base.py#L186-L222","documentation":"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.","triggerScenarios":"Batches where some records contain optional fields others lack (e.g. {'text': ...} vs {'text': ..., 'metadata': ...}), or rows built dynamically from heterogeneous sources before MLTransform.","commonSituations":"Joining documents from multiple sources into one PCollection; optional metadata fields present in only some records; schema drift between pipeline stages.","solutions":["Normalize all records to the same schema before batching (add None/default values for missing keys, drop extras).","Use a beam.Map to project each record onto a fixed key set matching the transform's columns.","Fix the upstream producer so it always emits the same fields."],"exampleFix":"// before\nbeam.Create([{'text': 'a'}, {'text': 'b', 'meta': 'm'}])\n// after\nbeam.Create([{'text': 'a', 'meta': None}, {'text': 'b', 'meta': 'm'}])","handlingStrategy":"validation","validationCode":"keys = {frozenset(d.keys()) for d in batch}\nassert len(keys) == 1, f'Heterogeneous batch keys: {keys}'","typeGuard":"def has_uniform_keys(batch) -> bool:\n    if not batch:\n        return True\n    expected = set(batch[0].keys())\n    return all(set(d.keys()) == expected for d in batch if isinstance(d, dict))","tryCatchPattern":"try:\n    out = data | MLTransform(...)\nexcept RuntimeError as e:\n    if 'same keys' in str(e):\n        raise ValueError(f'Upstream schema drift detected: {e}') from e\n    raise","preventionTips":["Define one canonical schema dict and project every record through it before MLTransform","Fill optional fields with None instead of omitting keys","Add schema checks in tests for upstream producers"],"tags":["python","apache-beam","schema"],"backgroundTag":"schema-validation-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}