apache/beam · error · KeyError

Enrichment requests to Vertex AI Feature Store should…

Error message

Enrichment requests to Vertex AI Feature Store should contain a field: %s in the input `beam.Row` to join the input with fetched response. This is used as the `FeatureViewDataKey` to fetch feature values corresponding to this key.

What it means

Raised by VertexAIFeatureStoreEnrichment.__call__ when the input beam.Row does not contain the configured row_key field. The key is required to build the FeatureViewDataKey used to look up feature values in Vertex AI Feature Store, so enrichment cannot proceed without it.

Solutions

Ensure every input beam.Row contains a field named exactly like the row_key passed to VertexAIFeatureStoreEnrichment.
Fix the row_key argument to match the actual column name in your input PCollection.
Add a Select() before enrichment to guarantee the key column is present and correctly named.

Example fix

// before
rows = pcoll | beam.Map(lambda x: beam.Row(name=x['name']))
result = rows | Enrichment(VertexAIFeatureStoreEnrichment('store', 'view', 'user_id'))
// after
rows = pcoll | beam.Map(lambda x: beam.Row(user_id=x['id'], name=x['name']))
result = rows | Enrichment(VertexAIFeatureStoreEnrichment('store', 'view', 'user_id'))

Defensive patterns

Strategy: validation

Validate before calling

def has_key(row, key):
    return key in row._asdict()
# before enrichment:
# assert all(has_key(r, 'user_id') for r in pcoll) or add Select(['user_id', ...])

Type guard

def key_present(row, key: str) -> bool:
    return hasattr(row, key)

Try / catch

try:
    result = rows | Enrichment(VertexAIFeatureStoreEnrichment('fs', 'fv', 'user_id'))
except KeyError as e:
    logging.error('Enrichment input missing key field: %s', e)

Prevention

Pin the input schema with Select() immediately before Enrichment.
Keep the row_key string in a shared constant used by both producers and the handler.

When it happens

Trigger: Calling the enrichment transform (DoFn __call__) with a beam.Row whose _asdict() has no entry for self.row_key — e.g. the Row was built without the entity-id column that was passed as row_key to the handler constructor.

Common situations: Renaming or dropping the key column in an upstream Select/Create, constructing beam.Row() manually and forgetting the key field, or passing a row_key string that doesn't match any column name (case/typo).

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/enrichment_handlers/vertex_ai_feature_store.py:157

    """Connect with the Vertex AI Feature Store."""
    self.client = aiplatform.gapic.FeatureOnlineStoreServiceClient(
        **self.kwargs)
    self.feature_view_path = self.client.feature_view_path(
        self.project,
        self.location,
        self.feature_store_name,
        self.feature_view_name)

  def __call__(self, request: beam.Row, *args, **kwargs):
    """Fetches feature value for an entity-id from Vertex AI Feature Store.

    Args:
      request: the input `beam.Row` to enrich.
    """
    try:
      entity_id = request._asdict()[self.row_key]
    except KeyError:
      raise KeyError(
          "Enrichment requests to Vertex AI Feature Store should "
          "contain a field: %s in the input `beam.Row` to join "
          "the input with fetched response. This is used as the "
          "`FeatureViewDataKey` to fetch feature values "
          "corresponding to this key." % self.row_key)
    try:
      response = self.client.fetch_feature_values(
          request=aiplatform.gapic.FetchFeatureValuesRequest(
              data_key=aiplatform.gapic.FeatureViewDataKey(key=entity_id),
              feature_view=self.feature_view_path,
              data_format=aiplatform.gapic.FeatureViewDataFormat.PROTO_STRUCT,
          ))
    except NotFound:
      if self.exception_level == ExceptionLevel.WARN:
        _LOGGER.warning(
            _not_found_err_message(
                self.feature_store_name, self.feature_view_name, entity_id))
        return request, beam.Row()

View on GitHub (pinned to 12126d8942)