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

expand-time guard in the Recommendations AI write/predict transform: neither the constructor's project argument nor the 'project' pipeline option was set, so the API client cannot be built against any GCP project.

Solutions

  1. Provide project=<project_id> when constructing the transform
  2. Set the project pipeline option: --project=<project_id>
Defensive patterns

Strategy: validation

When it happens

Trigger: Thrown at sdks/python/apache_beam/ml/gcp/recommendations_ai.py:232 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/b5d302a4cf4efc71. Report an issue: GitHub.

Appendix: source

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

            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.max_batch_size = max_batch_size
    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')
    return (
        pcoll | GroupIntoBatches.WithShardedKey(self.max_batch_size) | ParDo(
            _ImportCatalogItemsFn(
                self.project,
                self.retry,
                self.timeout,
                self.metadata,
                self.catalog_name)))


class _ImportCatalogItemsFn(DoFn):
  def __init__(
      self,
      project=None,
      retry=None,
      timeout=120,
      metadata=None,

View on GitHub (pinned to 12126d8942)