pyroid.integrations¶

Bases: UserWarning

patch replaced a module with pyroid's drop-in.

Says what was patched, which version pyroid reproduces and how to undo it; shown once per process for each name, plus once more if the installed package is another version. Silence it with warnings.simplefilter("ignore", pyroid.integrations.PatchWarning).

datatrove's FTFYFormatter (same arguments and defaults) on pyroid.ftfy.

doc.text becomes ftfy.fix_text(doc.text, config=self.config), byte for byte, computed batch_size documents at a time.

format(text: str) -> str

run(
    data: Iterable[Any], rank: int = 0, world_size: int = 1
) -> Any

datatrove's LanguageFilter (same arguments) on pyroid.fasttext.

Extra keyword-only arguments: model_path (a local lid.176.bin or GlotLID .bin; default: download to datatrove's cache, as datatrove does), batch_size and threads.

The decision and the language / language_score (and language_script, top_language_*_score) metadata come from datatrove's own filter() code. When only the top label can matter (no keep_top_pairs_threshold, and no languages, label_only, or a threshold above 0.501, since fastText's probabilities sum to at most 1 + 176e-5), pyroid predicts the top label only instead of all of them.

filter(doc: Any) -> Any

filter_batch(batch: list[Any]) -> list[Any]

datatrove's GopherRepetitionFilter (same arguments) on pyroid.filters.

datatrove's GopherQualityFilter (same arguments) on pyroid.filters.

datatrove's C4QualityFilter (same arguments) on pyroid.filters.

Kept documents get the same line removal; per-line statistics are not counted.

datatrove's FineWebQualityFilter (same arguments) on pyroid.filters.

datatrove's MinhashDedupSignature (stage 1) on pyroid.dedup.

Writes the same bucket_XXX/RANK.minhash.sig files, byte for byte (the sort pass is datatrove's own), so datatrove's later MinHash stages run on them unchanged. spaCy is not needed. language is datatrove's: shingles come from that language's word tokenizer, in Rust for every code pyroid tokenizes natively; another one runs datatrove's own step (pyroid.FallbackWarning). Note that datatrove's sort pass logs "Found existing sig files ... Skipping sig writing step": pyroid has just written them.

run(
    data: Iterable[Any], rank: int = 0, world_size: int = 1
) -> None

datatrove's PIIFormatter (same arguments and defaults) on pyroid.filters.

doc.text gets the same email and IP replacements, byte for byte, in the same rotation: the counters are datatrove's own (emails_replacer._replace_i and ip_replacer._replace_i), and whether an address is replaced is asked of datatrove's own validator, once per distinct address and batch. The regexes run in Rust, batch_size documents at a time.

format(text: str) -> str

run(
    data: Iterable[Any], rank: int = 0, world_size: int = 1
) -> Any

datatrove's C4BadWordsFilter (same arguments) on pyroid.filters.

The word search runs in Rust, batch_size documents at a time; the lists are loaded exactly as datatrove loads them (its download cache and pinned URLs). Everything else is datatrove's own: self.uniform (its default_rng(seed)) is drawn once per flagged document in order, and the same statistics are counted. A batch that contains a document whose language has no list raises before any of the batch is yielded (datatrove yields the documents before it first).

filter(doc: Any) -> Any

filter_batch(batch: list[Any]) -> list[Any]

datatrove's URLFilter (same arguments) on pyroid.filters.URLFilter.

The same keep/drop decisions and filter_reason strings (domain, subdomain, url, hard_blacklisted, soft_blacklisted, blacklisted_subword), from doc.metadata["url"], batch_size URLs per pyroid call. The suffix lists are this step's own tldextractor's (its cached Public Suffix List or tldextract's snapshot), and with use_integrated_lists the block lists are datatrove's, extracted into datatrove's cache as datatrove does; both are loaded at the first batch, in the worker. pyahocorasick is not needed.

What differs: a document without a url raises datatrove's AssertionError when its batch is filtered (before any document of that batch is passed on); datatrove's per-document DeprecationWarning from tldextract's registered_domain (hidden by Python's default filters) is not emitted; block_listed_domains and the other list attributes hold the constructor's entries only (the integrated lists live in Rust); banned_subwords_automaton is None.

download_data() -> None

filter(doc: Any) -> Any

filter_batch(batch: list[Any]) -> list[Any]

datatrove's SymbolLinesFormatter (same arguments) on pyroid.filters.

doc.text becomes the same text, byte for byte, computed batch_size documents at a time (FineWeb-2 runs it as SymbolLinesFormatter(symbols_to_remove=["|"], replace_char="\n")).

format(text: str) -> str

run(
    data: Iterable[Any], rank: int = 0, world_size: int = 1
) -> Any

datatrove's SentenceDedupSignature (stage 1) on pyroid.dedup.

Writes the same {bucket:04d}/{rank:05d}.c4_sig files, byte for byte (the same buckets, the same empty files), so datatrove's own later stages run on them unchanged. spaCy is not needed. language is datatrove's: every code pyroid tokenizes natively (pyroid.filters.languages()) runs in Rust; another one with split_sentences=True runs datatrove's own code (pyroid.FallbackWarning).

run(
    data: Iterable[Any], rank: int = 0, world_size: int = 1
) -> None

datatrove's SentenceFindDedups (stage 2) as a merge in Rust.

Reads the same signature (and index) files and writes the same {rank:04d}/{stem}.c4_dup files, byte for byte; local files are read by Rust directly, remote ones through fsspec. batch_size and threads are accepted for uniformity (the merge is one sequential pass).

run(
    data: Any = None, rank: int = 0, world_size: int = 1
) -> None

datatrove's SentenceDedupFilter (stage 3) on pyroid.dedup.

Reads the same .c4_dup files and yields the same documents with the same texts; the exclusion writer gets the same documents (>>> / <<<-marked texts). Only documents with duplicates go to Rust, batch_size at a time. spaCy is not needed. language is datatrove's: a code pyroid does not tokenize natively (pyroid.filters.languages()) that needs the tokenizer runs datatrove's own code (pyroid.FallbackWarning).

run(
    data: Iterable[Any], rank: int = 0, world_size: int = 1
) -> Any

datatrove's SentenceDedupBuildIndex as a merge in Rust (the same index file).

Like datatrove's own step, it fails (StopIteration) on an empty signature file: that case runs datatrove's code. batch_size and threads are accepted for uniformity.

run(
    data: Any = None, rank: int = 0, world_size: int = 1
) -> None

datatrove's Trafilatura extractor (same arguments and defaults) on pyroid.trafilatura.

doc.text becomes trafilatura.extract(doc.text, favor_precision=favour_precision, include_comments=False, deduplicate=deduplicate, **kwargs), byte for byte, computed in this process batch_size documents at a time; the same documents are forwarded and dropped, with the same statistics (total, extracted, forwarded, dropped, doc_len). Errors count as datatrove counts them: a page whose extraction raises is broken_process (it kills datatrove's child process, and the next child starts with an empty dedup cache, so the rest of the task does here too), and arguments that make trafilatura.extract("") raise (datatrove's warm-up call) make every page timeout. Trafilatura itself is not needed.

Extra keyword-only arguments: compat, the trafilatura release to reproduce (None: the installed trafilatura's version when pyroid reproduces it, which includes the 1.11.0 that datatrove[processing] installs, or pyroid's default, 2.3.1, when trafilatura is not installed; for any other installed version the documents go to datatrove's own step, with a one-time pyroid.FallbackWarning), batch_size and threads.

What differs: there is no child process and no timeout (pages datatrove gives up on after timeout seconds, counting timeout and restarting its dedup cache, get their text here). With deduplicate=True the dedup cache is a fresh pyroid.trafilatura.LRUCache per run, i.e. per task, as datatrove's per-task child process has a fresh one. Time is tracked per batch.

extract(text: str) -> str

run(
    data: Iterable[Any], rank: int = 0, world_size: int = 1
) -> Any

datatrove's WarcReader (same arguments and defaults) on pyroid.warc.iter_documents.

Each file is read as datatrove's read_file reads it, with warcio's parsing, the decompression and process_record in Rust: the same documents (text, id, metadata: url, date, file_path and default_metadata), the same records skipped, and the same exceptions, warnings and stderr messages at the same document, for gzip and plain files, on the local filesystem and on any other fsspec filesystem (ArchiveLoadFailed is warcio's class, as datatrove raises it, when warcio is installed). limit, skip, sharding and the statistics are datatrove's own code. The page text is decoded as datatrove decodes it: UTF-8, else cchardet's guess; python-magic is called only for records without WARC-Identified-Payload-Type. warcio itself is not needed.

How a file is read: a local .gz (compression="infer" or "gzip") or plain file is opened and read in Rust without the GIL, with the gzip layer reproducing CPython's gzip.GzipFile (datatrove reads through it); on another filesystem the file is opened with data_folder.open(path, "rb", compression=None) and that same emulation runs over it. Any other compression ("zstd", an inferred .bz2, ...), and .gz when fsspec's gzip opener is not CPython's (python-isal installed), use datatrove's own data_folder.open(..., compression=...), read as is.

What differs: time is tracked per document, including the reading of the records before it, plus once per file for the records after the last document (datatrove times each record's process_record).

read_file(filepath: str) -> Any

NeMo's ScoreFilter, scoring and filtering each batch in one pyroid call.

The same arguments as NeMo's, plus threads (keyword; default 1).

compute_filter_mask(
    df: DataFrame,
    filter_obj: Any,
    text_field: str,
    score_field: str | None,
    invert: bool,
) -> Series

NeMo's Score, scoring each batch in one pyroid call.

The same arguments as NeMo's, plus threads (keyword; default 1).

process(batch: Any) -> Any

NeMo's Filter, deciding each batch with array comparisons.

The same arguments as NeMo's, plus threads (keyword; default 1; unused, for symmetry).

compute_filter_mask(
    df: DataFrame,
    filter_fn: Any,
    filter_field: str,
    invert: bool,
) -> Series

NeMo's Modify, modifying each batch in one pyroid call.

The same arguments as NeMo's, plus threads (keyword; default 1). Modifiers with several input fields run NeMo's code.

process(batch: Any) -> Any

NeMo's FastTextLangId, with the model loaded by pyroid.fasttext.

The same arguments and the same score strings (str([score, lang])) as NeMo's.

model_check_or_download() -> None

load_model() -> None

score_document(text: str) -> str

keep_document(score: float | str) -> bool

NeMo's FastTextQualityFilter, with the model loaded by pyroid.fasttext.

The same arguments, scores and draws: keep_document draws from numpy's global random state (seeded once, when the filter is created), as NeMo does, also in pyroid's batched stages.

model_check_or_download() -> None

load_model() -> None

score_document(text: str) -> float

keep_document(score: float) -> bool

NeMo's UnicodeReformatter through pyroid.ftfy (ftfy itself is not needed).

The same arguments and defaults as NeMo's, which differ from ftfy's: no Unicode normalization, and ligatures, character widths, curly quotes and line breaks are left alone.

modify_document(text: str) -> str

What fineweb did.

Attributes:

• docs (int) –

  Documents in.

• kept (int) –

  Documents out.

• dropped (dict[str, int]) –

  Documents dropped per stage, in chain order (lid, the four filters, minhash; null too if there were null texts).

• reasons (dict[str, int]) –

  Documents dropped per reason ("gopher_quality:gopher_short_doc", "minhash", ...), most frequent first.

• seconds (dict[str, float]) –

  Wall time: annotate (the Dataset.map of ftfy, LID and the filters), dedup (MinHash over the survivors), total (including building the output datasets).

What fineweb returns.

Attributes:

• dataset (Dataset) –

  The kept documents, in order: cleaned text, the input's other columns, and language/language_score.

• annotated (Dataset) –

  Every document, with the text as it left the chain and the columns keep (the final decision) and reason (None if kept; "minhash" for a near-duplicate; see annotate).

• stats (Stats) –

  Counts and timings.

What FineWeb.annotate returns: one value per input document.

Attributes:

• text (ChunkedArray) –

  The text as it leaves the chain (after ftfy and C4's line removal); for a dropped document, the text the dropping stage saw.

• keep (ChunkedArray) –

  True if the document survived every stage but the dedup.

• reason (ChunkedArray) –

  None if kept, else "lid" or "<stage>:<datatrove reason>" (e.g. "gopher_quality:gopher_short_doc"); "null" for a null text.

• language (ChunkedArray) –

  fastText's top label, without __label__ (datatrove's language metadata).

• language_score (ChunkedArray) –

  Its probability (datatrove's language_score).

The FineWeb chain (see the module docstring), stage for stage as datatrove.

Parameters:

• lid_model (Any, default: None ) –

  A pyroid.fasttext.Model, a path to lid.176.bin, or None for lid_model_path's default (loaded on first use).

• languages (Sequence[str] | None, default: ('en',) ) –

  Languages to keep (None: any).

• language_threshold (float, default: 0.65 ) –

  datatrove LanguageFilter's language_threshold.

• ftfy_config (Any, default: None ) –

  A TextFixerConfig (pyroid.ftfy's or ftfy's); None is ftfy's default config, as in the benchmark chain. False skips ftfy.

• filters (Sequence[Any] | None, default: None ) –

  The quality filters, in order (pyroid.filters configs or datatrove filter instances); None is the FineWeb set above.

• minhash (Any, default: MINHASH_CONFIG ) –

  MinHash settings for new_dedup (a dict of MinhashConfig fields, or a MinhashConfig); None disables dedup.

• threads (int | None, default: None ) –

  Default threads= for every stage (None: the whole pool).

Language decisions are datatrove's: keep iff some requested language's probability is above the threshold. Above 0.501 that can only be the top label (fastText's probabilities sum to at most 1 + 176e-5), so the model is asked for its top label only; at lower thresholds it is asked for all labels.

model: Any

default_filters() -> list[Any]

stages() -> list[str]

new_dedup(
    *, threads: int | None = None
) -> MinHashDedup | None

fix(
    texts: Any, *, threads: int | None = None
) -> ChunkedArray

annotate(
    texts: Any, *, threads: int | None = None
) -> Annotation

datatrove examples/fineweb.py from WARC files, stage for stage as datatrove 0.10.1.

Per WARC file (one datatrove task each):

WarcReader (pyroid.warc.iter_documents: warcio's parsing, datatrove's process_record)
-> URLFilter() -> Trafilatura(favour_precision=True), trafilatura 1.11.0
-> LanguageFilter(languages=["en"], language_threshold=0.65)
-> GopherRepetitionFilter() -> GopherQualityFilter()
-> C4QualityFilter(filter_no_terminal_punct=False) -> FineWebQualityFilter()

then MinHash dedup (MINHASH_CONFIG) over the survivors of every file, in file order, and PIIFormatter() on what is left, a fresh rotation per file. documents, annotate (with a new_task extractor per file), new_dedup and new_pii are the pieces; pyroid clean --recipe fineweb-warc runs them.

Parameters:

• lid_model (Any, default: None ) –

  As FineWeb's.

• threads (int | None, default: None ) –

  Default threads= for every stage (None: the whole pool).

• url_filter (Any, default: None ) –

  A pyroid.filters.URLFilter (default: URLFilter(), built at first use: it needs tldextract and datatrove's block lists).

• trafilatura (dict[str, Any] | None, default: None ) –

  trafilatura.extract arguments over FINEWEB_TRAFILATURA.

url_filter: URLFilter

stages() -> list[str]

documents(
    source: Any,
    *,
    compression: Literal["infer", "gzip"] | None = "infer",
    limit: int = -1,
) -> Iterator[Document]

new_task() -> TaskExtractor

new_dedup(
    *, threads: int | None = None
) -> MinHashDedup | None

new_pii() -> PIIReplacer

annotate(
    docs: Sequence[Mapping[str, Any]],
    task: TaskExtractor,
    *,
    threads: int | None = None,
) -> WarcAnnotation

What FineWebWarc.annotate returns: one value per input document.

Attributes:

• text (list[str]) –

  The text as it leaves the chain; for a dropped document, the text the dropping stage saw (the HTML for the URL filter and Trafilatura).

• keep (list[bool]) –

  True if the document survived every stage but MinHash.

• reason (list[str | None]) –

  None if kept, else "url_filter:<reason>", "trafilatura" (no text extracted), "trafilatura:broken_process" (the extraction raised), "lid" or "<stage>:<datatrove reason>".

• language (list[str | None]) –

  fastText's top label (None before the LID stage).

• language_score (list[float | None]) –

  Its probability.

datatrove's Trafilatura step over the pages of one task, on pyroid.trafilatura.

Make one per datatrove task (in FineWeb, one per WARC file): it starts with an empty dedup cache, as the step's child process does. Called on a list of HTML pages, it returns for each the extracted text (None or "" when there is none: the step drops those), or the exception its extraction raised. datatrove's child process dies on such a page (the step counts broken_process and drops it) and the next child starts with an empty cache, so the cache is emptied after it here too. Pages are extracted in order: the dedup cache depends on it.

Parameters:

• threads (int | None, default: None ) –

  threads= of pyroid.trafilatura.extract_batch (None: the whole pool; with deduplication the pages are still extracted in order; threads parse ahead).

• **kwargs (Any, default: {} ) –

  trafilatura.extract arguments over FINEWEB_TRAFILATURA. With compat="1.11.0" the dedup cache is pyroid.trafilatura.LRU_TEST, as trafilatura 1.11.0 has only its module-level cache; with "2.3.1", an LRUCache of the task.

reset() -> None

__call__(htmls: Sequence[str]) -> list[Any]