apache/beam · error · RuntimeError

Data does not contain the following columns

Error message

Data does not contain the following columns : {missing_columns}.

What it means

Raised from the batch-processing helper behind MLTransform's dict columns: one or more column names listed for the transform are absent from the input dicts, so the requested per-column operation has nothing to apply to. The missing column names are interpolated into the message.

Solutions

1. Align the columns argument with the actual keys in your dicts (or rename dict keys to match columns).
2. Add a beam.Map to populate missing columns with defaults before the transform.
3. Print/log a sample element's keys to verify names before configuring columns.

Example fix

// before
MLTransform(...).with_transform(t(columns=['sentence']))
// after
MLTransform(...).with_transform(t(columns=['text']))  # 'text' is the real key

required = {'text'}
sample = next(iter(pcoll), None)
assert sample is None or required <= set(sample.keys()), f'Missing columns: {required - set(sample.keys())}'

def has_columns(d, columns) -> bool:
    return isinstance(d, dict) and set(columns) <= set(d.keys())

try:
    out = data | MLTransform(...)
except RuntimeError as e:
    if 'does not contain the following columns' in str(e):
        missing = str(e).split(': ')[-1]
        data = data | beam.Map(lambda d, m=missing.strip("{}"): {k.strip(" '"): None for k in m.split(',') if k.strip(" '") not in d} | d)
        out = data | MLTransform(...)

When it happens

Trigger: Specifying columns=['embedding_input'] while the input dicts only have a 'text' key; typos in column names; transforms configured with columns that the upstream data never produces.

Common situations: Renaming fields upstream without updating MLTransform(columns=...); copying example code whose column names differ from your data schema.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• A schema is required to write non-schema'd data.
• All dicts in batch must have the same keys. extra keys
• 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

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

Appendix: source

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

    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],
    batch: Sequence[Union[dict[str, Any], beam.Row]],
    embeddings: Sequence[Any]) -> list[Union[dict[str, Any], beam.Row]]:
  """Map embeddings back to columns in batch."""
  is_beam_row = False
  if batch and hasattr(batch[0], '_asdict'):
    is_beam_row = True

View on GitHub (pinned to 12126d8942)