{"record":{"id":"9bd373c03d9352eb","repo":"apache/beam","slug":"subclasses-must-implement-get-splitter-transform","errorCode":null,"errorMessage":"Subclasses must implement get_splitter_transform","messagePattern":"Subclasses must implement get_splitter_transform","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/ml/rag/chunking/base.py","lineNumber":76,"sourceCode":"      ...     chunks = (\n      ...         p \n      ...         | beam.Create([{'text': 'document...', 'source': 'doc.txt'}])\n      ...         | MLTransform(...).with_transform(chunker))\n\n    Args:\n      chunk_id_fn: Optional function to generate chunk IDs. If not provided,\n        random UUIDs will be used. Function should take a Chunk and return str.\n    \"\"\"\n    self.assign_chunk_id_fn = functools.partial(\n        _assign_chunk_id, chunk_id_fn) if chunk_id_fn is not None else None\n\n  @abc.abstractmethod\n  def get_splitter_transform(\n      self\n  ) -> beam.PTransform[beam.PCollection[dict[str, Any]],\n                       beam.PCollection[Chunk]]:\n    \"\"\"Creates transforms that emits splits for given content.\"\"\"\n    raise NotImplementedError(\n        \"Subclasses must implement get_splitter_transform\")\n\n  def get_ptransform_for_processing(\n      self, **kwargs\n  ) -> beam.PTransform[beam.PCollection[dict[str, Any]],\n                       beam.PCollection[Chunk]]:\n    \"\"\"Creates transform for processing documents into chunks.\"\"\"\n    ptransform = (\n        \"Split document\" >>\n        self.get_splitter_transform().with_output_types(Chunk))\n    if self.assign_chunk_id_fn:\n      ptransform = (\n          ptransform | \"Assign chunk id\" >> beam.Map(\n              self.assign_chunk_id_fn).with_output_types(Chunk))\n    return ptransform\n","sourceCodeStart":58,"sourceCodeEnd":92,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/ml/rag/chunking/base.py#L58-L92","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"# before\nclass MyChunker(RAGChunker):\n  def __init__(self, splitter):\n    self.splitter = splitter\n\n# after\nclass MyChunker(RAGChunker):\n  def __init__(self, splitter):\n    self.splitter = splitter\n\n  def get_splitter_transform(self):\n    return self.splitter  # a beam.PTransform emitting Chunks","handlingStrategy":"validation","validationCode":"from apache_beam.ml.rag.chunking.base import RAGChunker\n\ndef validate_chunker(chunker):\n    if not isinstance(chunker, RAGChunker):\n        raise TypeError('not a RAGChunker')\n    cls = type(chunker)\n    if cls.get_splitter_transform is RAGChunker.get_splitter_transform:\n        raise TypeError(f'{cls.__name__} must implement get_splitter_transform')\n    return True","typeGuard":"def is_concrete_chunker(obj) -> bool:\n    from apache_beam.ml.rag.chunking.base import RAGChunker\n    return (isinstance(obj, RAGChunker)\n            and type(obj).get_splitter_transform is not RAGChunker.get_splitter_transform)","tryCatchPattern":"try:\n    ptransform = chunker.get_ptransform_for_processing()\nexcept NotImplementedError as e:\n    logging.error('Chunker %s missing implementation: %s', type(chunker).__name__, e)\n    raise","preventionTips":["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."],"tags":["python","apache-beam","abstract-method","rag","chunking"],"backgroundTag":"abstract-method-not-implemented","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}