Implementation:Huggingface Datatrove BaseExtractor
| Knowledge Sources | |
|---|---|
| Domains | Text Extraction, Software Architecture |
| Last Updated | 2026-02-14 17:00 GMT |
Overview
Defines the abstract base class BaseExtractor and the ExtractorSandbox process isolation mechanism for safely extracting text from HTML or other non-plain text formats with timeout protection.
Description
BaseExtractor is the abstract base class for all text extraction pipeline steps in datatrove. It extends PipelineStep and requires subclasses to implement an extract(text: str) -> str method that converts non-plain text (typically HTML) into clean plain text. The base class handles the orchestration: iterating over documents, delegating extraction to a sandboxed subprocess, tracking statistics (total, extracted, timeout, broken_process, clean_error), and yielding successfully extracted documents.
The ExtractorSandbox class provides fault-tolerant process isolation for extraction operations. It spawns a daemon Process that runs the extraction function in a child process, communicating with the parent through a multiprocessing.Pipe. This design isolates the parent pipeline from crashes, memory leaks, and hangs in extraction libraries. Key safety features include:
- Timeout protection: Each document has a configurable extraction timeout (default 1 second for BaseExtractor, 0.1 seconds for ReadabilityInscriptis). The sandbox polls the pipe with a deadline and raises TimeoutError if the child does not respond in time.
- OOM protection: On Linux, the child process sets oom_score_adj to 1000, making it the first process killed by the OOM killer rather than the parent pipeline.
- Automatic recovery: When a child process dies (OOM-killed or crashed), the sandbox detects this via process.is_alive() checks and spawns a new child for the next document.
- Comprehensive cleanup: The __exit__ method terminates, joins, and if necessary kills all spawned processes to prevent zombie processes.
A warmup text is sent to the child process during initialization to trigger lazy imports and model loading before the timeout-protected processing begins.
Usage
Subclass BaseExtractor to create custom text extractors. Implement the extract method with your extraction logic. The sandbox, timeout handling, statistics tracking, and document pipeline iteration are all handled automatically by the base class.
Code Reference
Source Location
- Repository: Huggingface_Datatrove
- File: src/datatrove/pipeline/extractors/base.py
- Lines: 1-222
Signature
class BaseExtractor(PipelineStep):
type = "🛢 - EXTRAC"
@abstractmethod
def __init__(self, timeout: float = 1): ...
@abstractmethod
def extract(self, text: str) -> str: ...
def run(self, data: DocumentsPipeline, rank: int = 0, world_size: int = 1) -> DocumentsPipeline: ...
class ExtractorSandbox:
def __init__(self, timeout, warmup_text=""): ...
def process_document(self, text, extract_fn): ...
def __enter__(self): ...
def __exit__(self, exc_type, exc_val, exc_tb): ...
Import
from datatrove.pipeline.extractors.base import BaseExtractor, ExtractorSandbox
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| timeout | float | No | Maximum seconds for extracting a single document (default: 1) |
| data | DocumentsPipeline | Yes | Generator of Document objects with non-plain text in doc.text |
| rank | int | No | Task rank (default: 0) |
| world_size | int | No | Total number of tasks (default: 1) |
Outputs
| Name | Type | Description |
|---|---|---|
| DocumentsPipeline | Generator[Document] | Documents with doc.text replaced by extracted plain text; empty or failed extractions are dropped |
| Statistics | Stats | Counts of total, extracted, timeout, broken_process, clean_error, forwarded, dropped |
Usage Examples
Basic Usage
from datatrove.pipeline.extractors.base import BaseExtractor
class MyCustomExtractor(BaseExtractor):
"""Custom extractor that strips HTML tags using a simple regex."""
def __init__(self, timeout: float = 1):
super().__init__(timeout)
def extract(self, text: str) -> str:
import re
# Simple HTML tag removal (for demonstration only)
clean = re.sub(r'<[^>]+>', '', text)
return clean.strip()
# Use in a pipeline
from datatrove.executor.local import LocalPipelineExecutor
executor = LocalPipelineExecutor(
pipeline=[
# ... reader step that provides HTML documents ...
MyCustomExtractor(timeout=2.0),
# ... writer step ...
],
tasks=1,
logging_dir="logs/extraction",
)
executor.run()