apache/beam · error · NotImplementedError

NotImplementedError(type(self))

Error message

NotImplementedError(type(self))

What it means

VectorDatabaseWriteConfig.create_write_transform is an abstract-style hook: subclasses must return a beam.PTransform that writes EmbeddableItems to the target vector DB. The base class raises NotImplementedError(type(self)) when a caller uses a config subclass that never implemented this method.

Source

Thrown at sdks/python/apache_beam/ml/rag/ingestion/base.py:60

    ...     def create_write_transform(self):
    ...         return beam.io.WriteToBigQuery(
    ...             table=self.table
    ...         )
  """
  @abstractmethod
  def create_write_transform(self) -> beam.PTransform[EmbeddableItem, Any]:
    """Creates a PTransform that writes embeddings to the vector database.

    Returns:
        A PTransform that accepts PCollection[EmbeddableItem]
        and writes the embeddings
        and metadata to the configured vector database.
        The transform should handle:
        - Converting EmbeddableItem format to database schema
        - Setting up database connection/client
        - Writing with appropriate batching/error handling
    """
    raise NotImplementedError(type(self))


class VectorDatabaseWriteTransform(beam.PTransform):
  """A PTransform for writing embedded chunks to vector databases.
  
  This transform uses a VectorDatabaseWriteConfig to write chunks with
  embeddings to vector database. It handles validating the config and applying
  the database-specific write transform.

  Example usage:
    >>> config = BigQueryVectorConfig(
    ...     table='project.dataset.embeddings',
    ...     embedding_column='embedding'
    ... )
    >>>
    >>> with beam.Pipeline() as p:
    ...     items = p | beam.Create([...])  # PCollection[EmbeddableItem]
    ...     items | VectorDatabaseWriteTransform(config)

View on GitHub (pinned to 12126d8942)

Solutions

1. Override create_write_transform in your config subclass and return a beam.PTransform that writes to your database
2. Use one of the built-in configs (e.g., BigQueryVectorWriterConfig) instead of the base class
3. Never instantiate VectorDatabaseWriteConfig directly

Example fix

// before
class MyConfig(VectorDatabaseWriteConfig):
    pass  # no create_write_transform
// after
class MyConfig(VectorDatabaseWriteConfig):
    def create_write_transform(self):
        return _WriteToMyVectorDb(self)

import inspect
assert not inspect.isabstract(type(cfg)) and type(cfg).create_write_transform is not VectorDatabaseWriteConfig.create_write_transform

def has_write_transform(cfg) -> bool:
    return type(cfg).create_write_transform is not VectorDatabaseWriteConfig.create_write_transform

try:
    transform = cfg.create_write_transform()
except NotImplementedError:
    transform = default_bigquery_write_transform()

When it happens

Trigger: Instantiating a custom VectorDatabaseWriteConfig subclass (or the bare base class) and passing it to VectorDatabaseWriteTransform, whose expand() calls create_write_transform().

Common situations: Writing a custom DB integration but forgetting to override create_write_transform; using the abstract base directly in tests; a subclass that only overrode other hooks.

Understand the failure class

Background: "NotImplementedError: Subclasses should override this method" / "must be implemented" — abstract method errors explained — this error's family across 40 libraries.

Related errors

• Subclasses must implement get_splitter_transform
• database_config must be VectorDatabaseWriteConfig, got {type
• Encode not implemented: %s.
• Decode not implemented: %s.
• NotImplementedError

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