apache/beam · error · KeyError

row_key not found in input PCollection.

Error message

row_key %s not found in input PCollection.

What it means

The BigTable enrichment handler wraps a KeyError from looking up the requested row key and re-raises it with a clear message: the row_key value used for the lookup was not found in the input PCollection's fields. This typically means the enrichment input row lacks the field named by row_key (or the field value is missing), so the key extraction failed before BigTable was even queried.

Solutions

  1. Ensure every input row has a non-null field matching the handler's row_key name
  2. Update the handler's row_key configuration to the correct column name
  3. Add a validation/filter transform before enrichment to drop or repair rows missing the key

Example fix

// before
BigTableEnrichmentHandler(row_key='user_id', ...)
# input rows only have 'userId'
// after
BigTableEnrichmentHandler(row_key='userId', ...)
# or: beam.Map(lambda r: r | {'userId': r['user_id']}) first
Defensive patterns

Strategy: validation

Validate before calling

def row_has_key(row, key):
    return key in row._asdict() and row._asdict()[key] is not None
# assert all(row_has_key(r, 'userId') for r in rows) before enriching

Try / catch

try:
    enriched = beam_rows | beam.transforms.enrichment.Enrichment(handler)
except KeyError as e:
    if 'row_key' in str(e):
        raise ValueError('input rows missing configured row_key field') from e
    raise

Prevention

When it happens

Trigger: Using enrichment_handlers.BigTableEnrichmentHandler where the input Beam.Row does not contain the configured row_key field, or row_key_str is absent/None so response_dict key access raises KeyError.

Common situations: Renaming columns upstream without updating the handler's row_key config; events that optionally lack the key field; typos in field names.

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/enrichment_handlers/bigtable.py:147

          response_dict[cf_id] = {}
          for col_id, col_v in cf_v.items():
            if self._include_timestamp:
              response_dict[cf_id][col_id.decode(self._encoding)] = [
                  (v.value.decode(self._encoding), v.timestamp) for v in col_v
              ]
            else:
              response_dict[cf_id][col_id.decode(
                  self._encoding)] = col_v[0].value.decode(self._encoding)
      elif self._exception_level == ExceptionLevel.WARN:
        _LOGGER.warning(
            'no matching row found for row_key: %s '
            'with row_filter: %s' % (row_key_str, self._row_filter))
      elif self._exception_level == ExceptionLevel.RAISE:
        raise ValueError(
            'no matching row found for row_key: %s '
            'with row_filter=%s' % (row_key_str, self._row_filter))
    except KeyError:
      raise KeyError('row_key %s not found in input PCollection.' % row_key_str)
    except NotFound:
      raise NotFound(
          'GCP BigTable cluster `%s:%s:%s` not found.' %
          (self._project_id, self._instance_id, self._table_id))
    except Exception as e:
      raise e

    return request, beam.Row(**response_dict)

  def __exit__(self, exc_type, exc_val, exc_tb):
    """Clean the instantiated BigTable client."""
    self.client = None
    self.instance = None
    self._table = None

  def get_cache_key(self, request: beam.Row) -> str:
    """Returns a string formatted with row key since it is unique to
    a request made to `Bigtable`."""

View on GitHub (pinned to 12126d8942)