From ff23b944eca5b508d9874c01aef3395bd5bd847e Mon Sep 17 00:00:00 2001 From: guipenedo Date: Mon, 4 Mar 2024 17:22:08 +0100 Subject: [PATCH 1/9] bugfix mailuser --- src/datatrove/executor/slurm.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/datatrove/executor/slurm.py b/src/datatrove/executor/slurm.py index d06fe5820..5da97fb20 100644 --- a/src/datatrove/executor/slurm.py +++ b/src/datatrove/executor/slurm.py @@ -292,8 +292,7 @@ def get_sbatch_args(self, max_array: int = 1) -> dict: "array": f"0-{max_array - 1}{f'%{self.workers}' if self.workers != -1 else ''}", "requeue": "", "qos": self.qos, - "mail-type": self.mail_type, - "mail-user": self.mail_user, + **({"mail-type": self.mail_type, "mail-user": self.mail_user} if self.mail_user else {}), **self._sbatch_args, } From 6baacca1dd5b4e0c9c2ae9e830eabc3343f788f1 Mon Sep 17 00:00:00 2001 From: Leandro von Werra Date: Tue, 5 Mar 2024 12:09:10 +0100 Subject: [PATCH 2/9] Add `jobs_status` command. (#113) * add command file and project setting * fix style * display precentage and remove unused regex * show complete as well by default * small improvements --------- Co-authored-by: guipenedo --- pyproject.toml | 1 + src/datatrove/tools/jobs_status.py | 91 ++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+) create mode 100644 src/datatrove/tools/jobs_status.py diff --git a/pyproject.toml b/pyproject.toml index ffd028c66..35cf167af 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -85,6 +85,7 @@ merge_stats = "datatrove.tools.merge_stats:main" launch_pickled_pipeline = "datatrove.tools.launch_pickled_pipeline:main" failed_logs = "datatrove.tools.failed_logs:main" inspect_data = "datatrove.tools.inspect_data:main" +jobs_status = "datatrove.tools.jobs_status:main" [build-system] requires = ["setuptools"] diff --git a/src/datatrove/tools/jobs_status.py b/src/datatrove/tools/jobs_status.py new file mode 100644 index 000000000..dcfa80d2b --- /dev/null +++ b/src/datatrove/tools/jobs_status.py @@ -0,0 +1,91 @@ +import argparse +import json +import os.path + +from loguru import logger +from rich.console import Console + +from datatrove.io import get_datafolder +from datatrove.utils._import_utils import is_rich_available + + +if not is_rich_available(): + raise ImportError("Please install `rich` to run this command (`pip install rich`).") + + +parser = argparse.ArgumentParser("Fetch all jobs that are running or complete.") + +parser.add_argument( + "path", type=str, nargs="?", help="Path to the logging folder. Defaults to current directory.", default=os.getcwd() +) + +parser.add_argument( + "-p", "--log_prefix", type=str, nargs="?", help="Prefix of logging folders to be scanned.", default="" +) +parser.add_argument("-hc", "--hide_complete", help="Hide all jobs that are already complete.", action="store_true") + + +def main(): + """ + Takes a `path` as input, gets all valid job folders and their total number of tasks from `executor.json` and then gets which ranks are + incomplete by scanning `path/{LOGGING_DIRS}/completions`. If a `log_prefix` is provided the directories following the `path/log_prefix{LOGGING_DIRS}/completions` + pattern are scanned. + """ + args = parser.parse_args() + console = Console() + + main_folder = get_datafolder(args.path) + logging_dirs = [ + f + for f, info in main_folder.glob(f"{args.log_prefix}*", detail=True, maxdepth=1).items() + if info["type"] == "directory" + ] + logger.remove() + + complete_jobs = 0 + incomplete_jobs = 0 + + for path in logging_dirs: + logging_dir = get_datafolder(main_folder.resolve_paths(path)) + if not logging_dir.isfile("executor.json"): + console.log( + f'Could not find "executor.json" in the given directory ({path}). Are you sure it is a ' + "logging folder?", + style="red", + ) + continue + with logging_dir.open("executor.json", "rt") as f: + world_size = json.load(f).get("world_size", None) + if not world_size: + console.log( + f"Could not get the total number of tasks in {path}, please try relaunching the run.", + style="red", + ) + continue + + with console.status("Fetching list of incomplete tasks"): + completed = set(logging_dir.list_files("completions")) + incomplete = set(filter(lambda rank: f"completions/{rank:05d}" not in completed, range(world_size))) + + if len(incomplete) == 0: + emoji = "✅" + complete_jobs += 1 + else: + emoji = "❌" + incomplete_jobs += 1 + + if len(incomplete) > 0 or not args.hide_complete: + console.log( + f"{emoji} {path.split('/')[-1]+':': <50}{len(completed)}/{world_size} ({len(completed)/(world_size):.0%}) completed tasks." + ) + + if complete_jobs + incomplete_jobs > 0: + console.log( + f"Summary: {complete_jobs}/{complete_jobs+incomplete_jobs} ({complete_jobs/(complete_jobs+incomplete_jobs):.0%}) jobs completed." + ) + else: + console.log("No jobs found.") + + +if __name__ == "__main__": + main() From 366c4cc3a3c3b6054af891b69013562045559a3e Mon Sep 17 00:00:00 2001 From: guipenedo Date: Tue, 5 Mar 2024 12:10:19 +0100 Subject: [PATCH 3/9] bugfix recursive --- src/datatrove/io.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/datatrove/io.py b/src/datatrove/io.py index 5eb377612..8722ea394 100644 --- a/src/datatrove/io.py +++ b/src/datatrove/io.py @@ -139,11 +139,11 @@ def list_files( [ f for f, info in ( - self.find(subdirectory, maxdepth=0 if not recursive else None, detail=True, **extra_options) + self.find(subdirectory, maxdepth=1 if not recursive else None, detail=True, **extra_options) if not glob_pattern else self.glob( self.fs.sep.join([glob_pattern, subdirectory]), - maxdepth=0 if not recursive else None, + maxdepth=1 if not recursive else None, detail=True, **extra_options, ) From df0e324197a0f1c867d92a9929c8d5d7f45752b0 Mon Sep 17 00:00:00 2001 From: guipenedo Date: Tue, 5 Mar 2024 12:17:24 +0100 Subject: [PATCH 4/9] nit --- src/datatrove/tools/jobs_status.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/datatrove/tools/jobs_status.py b/src/datatrove/tools/jobs_status.py index dcfa80d2b..dc730d62b 100644 --- a/src/datatrove/tools/jobs_status.py +++ b/src/datatrove/tools/jobs_status.py @@ -76,7 +76,7 @@ def main(): if len(incomplete) > 0 or not args.hide_complete: console.log( - f"{emoji} {path.split('/')[-1]+':': <50}{len(completed)}/{world_size} ({len(completed)/(world_size):.0%}) completed tasks." + f"{emoji} {path + ':': <50}{len(completed)}/{world_size} ({len(completed)/(world_size):.0%}) completed tasks." ) if complete_jobs + incomplete_jobs > 0: From 7dbf8eccb0d41d365358a6d98bf83e078b7bd96d Mon Sep 17 00:00:00 2001 From: guipenedo Date: Tue, 5 Mar 2024 14:11:24 +0100 Subject: [PATCH 5/9] bugfix tokenization unshuffled cleanup --- src/datatrove/pipeline/tokens/tokenizer.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/datatrove/pipeline/tokens/tokenizer.py b/src/datatrove/pipeline/tokens/tokenizer.py index a5cd04d76..f226ff9fd 100644 --- a/src/datatrove/pipeline/tokens/tokenizer.py +++ b/src/datatrove/pipeline/tokens/tokenizer.py @@ -105,6 +105,8 @@ def cleanup(self): self.output_folder.rm_file(self.filename) if self.loss_file: self.output_folder.rm_file(f"{self.filename}.loss") + if self.save_final_metadata and self.output_folder.exists(f"{self.filename}.metadata"): + self.output_folder.rm_file(f"{self.filename}.metadata") def write_bytes(self, tk_bytes: bytes, doc_ends: list[int] = None): """Write tk_bytes to the tokens file and update the document boundaries with a new document end (in tokens). From 6a15cabf98c9e805fb0dc8ef822fbd11323a781a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mario=20=C5=A0a=C5=A1ko?= Date: Tue, 5 Mar 2024 20:46:22 +0100 Subject: [PATCH 6/9] Re-enable `datasets` test (#114) --- pyproject.toml | 2 +- src/datatrove/pipeline/readers/huggingface.py | 2 +- tests/pipeline/test_hf_reader.py | 13 +++++++------ 3 files changed, 9 insertions(+), 8 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 35cf167af..77cceb6c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,7 +41,7 @@ io = [ "pyarrow", "python-magic", "warcio", - "datasets" + "datasets>=2.18.0" ] s3 = [ "s3fs>=2023.12.2", diff --git a/src/datatrove/pipeline/readers/huggingface.py b/src/datatrove/pipeline/readers/huggingface.py index 36ea7e4a6..fca57cce6 100644 --- a/src/datatrove/pipeline/readers/huggingface.py +++ b/src/datatrove/pipeline/readers/huggingface.py @@ -52,7 +52,7 @@ def get_document_from_dict(self, data: dict, source: str, id_in_file: int | str) return document def run(self, data: DocumentsPipeline = None, rank: int = 0, world_size: int = 1) -> DocumentsPipeline: - from datasets import load_dataset + from datasets import load_dataset # type: ignore if data: yield from data diff --git a/tests/pipeline/test_hf_reader.py b/tests/pipeline/test_hf_reader.py index 8df74234f..2c4267ba9 100644 --- a/tests/pipeline/test_hf_reader.py +++ b/tests/pipeline/test_hf_reader.py @@ -1,14 +1,15 @@ import unittest +from datatrove.pipeline.readers import HuggingFaceDatasetReader + from ..utils import require_datasets @require_datasets class TestHuggingFaceReader(unittest.TestCase): def test_read_dataset(self): - # reader = HuggingFaceDatasetReader( - # "truthful_qa", dataset_options={"name": "generation", "split": "validation"}, text_key="question" - # ) - # data = list(reader()) - # assert len(data) == 817 - pass + reader = HuggingFaceDatasetReader( + "truthful_qa", dataset_options={"name": "generation", "split": "validation"}, text_key="question" + ) + data = list(reader()) + assert len(data) == 817 From e1f95246257abd71517885c10470b077cda96cf3 Mon Sep 17 00:00:00 2001 From: Zehan Li <69186130+jordane95@users.noreply.github.com> Date: Wed, 6 Mar 2024 20:51:50 +0800 Subject: [PATCH 7/9] Update warc.py (#115) Fix typo --- src/datatrove/pipeline/readers/warc.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/datatrove/pipeline/readers/warc.py b/src/datatrove/pipeline/readers/warc.py index e165f8ebc..6249c13fb 100644 --- a/src/datatrove/pipeline/readers/warc.py +++ b/src/datatrove/pipeline/readers/warc.py @@ -28,7 +28,7 @@ class WarcReader(BaseDiskReader): """ name = "🕷 Warc" - _requires_dependencies = ["warcio", ("cchardet", "faust-chardet"), ("magic", "python-magic")] + _requires_dependencies = ["warcio", ("cchardet", "faust-cchardet"), ("magic", "python-magic")] def __init__( self, From f63dae7d3df9ca16861fcfa6c730448186558b26 Mon Sep 17 00:00:00 2001 From: vsabolcec Date: Thu, 7 Mar 2024 08:46:30 +0100 Subject: [PATCH 8/9] add all test_gopher_quality() tests --- tests/pipeline/test_filters.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/pipeline/test_filters.py b/tests/pipeline/test_filters.py index ebbfae270..104b2a5d6 100644 --- a/tests/pipeline/test_filters.py +++ b/tests/pipeline/test_filters.py @@ -65,12 +65,20 @@ def test_gopher_repetition(self): def test_gopher_quality(self): gopher_quality = GopherQualityFilter(min_doc_words=10, max_doc_words=1000) self.check_filter(gopher_quality, get_doc("I am too small..."), "gopher_short_doc") + self.check_filter(gopher_quality, get_doc("Very long document. " * 400), "gopher_long_doc") self.check_filter(gopher_quality, get_doc("I am " * 20), "gopher_below_avg_threshold") self.check_filter(gopher_quality, get_doc("interconnection " * 20), "gopher_above_avg_threshold") self.check_filter(gopher_quality, get_doc("# comment " * 20), "gopher_too_many_hashes") self.check_filter(gopher_quality, get_doc("... comment " * 20), "gopher_too_many_ellipsis") + self.check_filter(gopher_quality, get_doc("• comment\n" * 20), "gopher_too_many_bullets") + self.check_filter( + gopher_quality, + get_doc("comment comment comment comment comment comment comment comment comment...\n" * 20), + "gopher_too_many_end_ellipsis", + ) text = "the ./!*?<><> apple orange ++ interconnection !<>??? have" * 20 self.check_filter(gopher_quality, get_doc(text), "gopher_below_alpha_threshold") + self.check_filter(gopher_quality, get_doc("No stopwords. " * 10), "gopher_enough_stop_words") self.assertTrue(gopher_quality(get_doc(TEXT_LF_1))) def test_lambda(self): From 398bc4c10a7f359732551e6450e9dc3b5076d69b Mon Sep 17 00:00:00 2001 From: vsabolcec Date: Thu, 7 Mar 2024 08:55:19 +0100 Subject: [PATCH 9/9] Add test_list() --- tests/pipeline/test_filters.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/pipeline/test_filters.py b/tests/pipeline/test_filters.py index 104b2a5d6..ff475e5e4 100644 --- a/tests/pipeline/test_filters.py +++ b/tests/pipeline/test_filters.py @@ -6,6 +6,7 @@ GopherRepetitionFilter, LambdaFilter, LanguageFilter, + ListFilter, RegexFilter, UnigramLogProbFilter, URLFilter, @@ -127,3 +128,10 @@ def test_url(self): assert url_filter.filter(doc) else: self.check_filter(url_filter, doc, result) + + def test_list(self): + list_filter = ListFilter() + self.check_filter(list_filter, get_doc("List\n" * 5), "Suspected list") + self.check_filter(list_filter, get_doc("Also list\n" * 5), "Suspected list") + self.check_filter(list_filter, get_doc("And another list\n" * 5), "Suspected list") + self.assertTrue(list_filter.filter(get_doc("Not a list anymore\n" * 5)) is True)