apache/beam · error · NotImplementedError

Subclasses must implement get_splitter_transform

Error message

Subclasses must implement get_splitter_transform

What it means

This NotImplementedError is raised by the abstract RAGChunker.get_splitter_transform method in apache_beam.ml.rag.chunking.base when a subclass fails to implement it. The base class defines the chunking pipeline contract: get_ptransform_for_processing composes the pipeline by calling get_splitter_transform, so any concrete chunker must supply a Beam PTransform that converts input documents into Chunks. Raising NotImplementedError instead of leaving @abc.abstractmethod alone catches subclasses that bypass proper instantiation checks.

Source

Thrown at sdks/python/apache_beam/ml/rag/chunking/base.py:76

      ...     chunks = (
      ...         p 
      ...         | beam.Create([{'text': 'document...', 'source': 'doc.txt'}])
      ...         | MLTransform(...).with_transform(chunker))

    Args:
      chunk_id_fn: Optional function to generate chunk IDs. If not provided,
        random UUIDs will be used. Function should take a Chunk and return str.
    """
    self.assign_chunk_id_fn = functools.partial(
        _assign_chunk_id, chunk_id_fn) if chunk_id_fn is not None else None

  @abc.abstractmethod
  def get_splitter_transform(
      self
  ) -> beam.PTransform[beam.PCollection[dict[str, Any]],
                       beam.PCollection[Chunk]]:
    """Creates transforms that emits splits for given content."""
    raise NotImplementedError(
        "Subclasses must implement get_splitter_transform")

  def get_ptransform_for_processing(
      self, **kwargs
  ) -> beam.PTransform[beam.PCollection[dict[str, Any]],
                       beam.PCollection[Chunk]]:
    """Creates transform for processing documents into chunks."""
    ptransform = (
        "Split document" >>
        self.get_splitter_transform().with_output_types(Chunk))
    if self.assign_chunk_id_fn:
      ptransform = (
          ptransform | "Assign chunk id" >> beam.Map(
              self.assign_chunk_id_fn).with_output_types(Chunk))
    return ptransform

View on GitHub (pinned to 12126d8942)

Solutions

Implement get_splitter_transform in your subclass, returning a beam.PTransform that maps PCollection[dict] to PCollection[Chunk] (e.g. reuse MLTransform or a custom DoFn).
If you only need standard splitters, use the provided chunkers (e.g. LangChainChunker) instead of subclassing RAGChunker yourself.
Check that the class you are actually instantiating is the concrete subclass, not the abstract base, and that no intermediate class in the MRO forgot the override.

Example fix

# before
class MyChunker(RAGChunker):
  def __init__(self, splitter):
    self.splitter = splitter

# after
class MyChunker(RAGChunker):
  def __init__(self, splitter):
    self.splitter = splitter

  def get_splitter_transform(self):
    return self.splitter  # a beam.PTransform emitting Chunks

Defensive patterns

Strategy: validation

Validate before calling

from apache_beam.ml.rag.chunking.base import RAGChunker

def validate_chunker(chunker):
    if not isinstance(chunker, RAGChunker):
        raise TypeError('not a RAGChunker')
    cls = type(chunker)
    if cls.get_splitter_transform is RAGChunker.get_splitter_transform:
        raise TypeError(f'{cls.__name__} must implement get_splitter_transform')
    return True

Type guard

def is_concrete_chunker(obj) -> bool:
    from apache_beam.ml.rag.chunking.base import RAGChunker
    return (isinstance(obj, RAGChunker)
            and type(obj).get_splitter_transform is not RAGChunker.get_splitter_transform)

Try / catch

try:
    ptransform = chunker.get_ptransform_for_processing()
except NotImplementedError as e:
    logging.error('Chunker %s missing implementation: %s', type(chunker).__name__, e)
    raise

Prevention

Always override every @abc.abstractmethod when subclassing RAGChunker.
Instantiate one instance in a unit test and call get_ptransform_for_processing before submitting pipelines.
Prefer built-in chunkers (LangChainChunker, etc.) over custom subclasses when possible.

When it happens

Trigger: Instantiating a subclass of RAGChunker that overrides get_ptransform_for_processing usage but does not override get_splitter_transform, then calling get_ptransform_for_processing (directly or by applying the chunker to a Beam pipeline, e.g. beam.ParDo(chunker.get_ptransform_for_processing())). Also occurs when a subclass explicitly calls super().get_splitter_transform() without implementing its own version.

Common situations: Writing a custom chunker class that inherits from RAGChunker (or LegacyChunker) but only implements __init__ and forgets get_splitter_transform; upgrading Apache Beam where a previously-used third-party chunker no longer implements the current interface; copy-pasting a skeleton subclass without filling in the abstract method.

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

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