apache/beam · error · ValueError

GCP project name needs to be specified in "project"…

Error message

GCP project name needs to be specified in "project" pipeline
            option

What it means

WriteUserEvent/WritePrediction expand-time guard: the transform's project argument is None and the pipeline's GoogleCloudOptions.project is also unset, so there is no GCP project in which to look up the Recommendations AI catalog/event store.

Solutions

  1. Pass project explicitly to the transform constructor
  2. Set --project on the pipeline options or via options.view_as(GoogleCloudOptions).project
Defensive patterns

Strategy: validation

When it happens

Trigger: Thrown at sdks/python/apache_beam/ml/gcp/recommendations_ai.py:126 when the library encounters an invalid state.

Common situations: See trigger scenarios.

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

Appendix: source

Thrown at sdks/python/apache_beam/ml/gcp/recommendations_ai.py:126

              errors, if any, should be retried.
            timeout (float): Optional. The amount of time, in seconds, to wait
              for the request to complete.
            metadata: Optional. Strings which
              should be sent along with the request as metadata.
            catalog_name (str): Optional. Name of the catalog.
              Default: 'default_catalog'
        """
    self.project = project
    self.retry = retry
    self.timeout = timeout
    self.metadata = metadata
    self.catalog_name = catalog_name

  def expand(self, pcoll):
    if self.project is None:
      self.project = pcoll.pipeline.options.view_as(GoogleCloudOptions).project
    if self.project is None:
      raise ValueError(
          """GCP project name needs to be specified in "project" pipeline
            option""")
    pardo = ParDo(
        _CreateCatalogItemFn(
            self.project,
            self.retry,
            self.timeout,
            self.metadata,
            self.catalog_name))
    return pcoll | pardo.with_outputs(
        FAILED_CATALOG_ITEMS, main='created_catalog_items')


class _CreateCatalogItemFn(DoFn):
  def __init__(
      self,
      project: str = None,
      retry: Retry = None,

View on GitHub (pinned to 12126d8942)