Repository: allenai/dolma (language: python)

Issue:
ff



Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index b0131293..c5829b08 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -682,7 +682,7 @@ dependencies = [
 
 [[package]]
 name = "dolma"
-version = "0.8.0"
+version = "0.9.0"
 dependencies = [
  "ahash",
  "aws-config",
diff --git a/Cargo.toml b/Cargo.toml
index 4a177cd3..41b403f0 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "dolma"
-version = "0.8.0"
+version = "0.9.0"
 edition = "2021"
 license = "Apache-2.0"
 
diff --git a/examples/README.md b/examples/README.md
new file mode 100644
index 00000000..a28e391f
--- /dev/null
+++ b/examples/README.md
@@ -0,0 +1,3 @@
+# Dolma Examples
+
+This directory contains examples on how to use Dolma to pre-process data for LLM training.
diff --git a/examples/c4-like/README.md b/examples/c4-like/README.md
new file mode 100644
index 00000000..c7cd541a
--- /dev/null
+++ b/examples/c4-like/README.md
@@ -0,0 +1,62 @@
+# C4-like processing
+
+## Step 0
+
+Convert data to dolma format. For example, we reformat the [trafilatura](https://trafilatura.readthedocs.io/en/latest/) output to dolma format.
+
+
+```bash
+python examples/c4-like/reformat_trafilatura.py \
+    src='s3://ai2-llm/experimental/dcnlp_beta_ai2/raw/cc_trafilatura_beta/crawl-data/*/segments/*/warc/*.jsonl' \
+    dst='s3://ai2-llm/experimental/dcnlp_beta_ai2/v0/documents' \
+    proc=16 \
+    debug=False
+```
+
+## Step 1
+
+Tag with fasttext language id and c4 rules.
+
+```bash
+time dolma -c examples/c4-like/tagger.yaml tag
+```
+
+Timing on a `c6a.4xlarge` instance (8 cores/16 threads, 32 GB memory, gp3 volume/16000 IOPS/125 MB/s throughput):
+
+| **Processes** | **Seconds** |
+|:-------------:|:-----------:|
+|       1       |     N/A     |
+|       8       |     655     |
+|      16       |     ???     |
+
+
+## Step 2
+
+Run mixer to generate the final dataset.
+
+```bash
+time dolma -c examples/c4-like/mixer.yaml mix
+```
+
+Timing on a `c6a.4xlarge` instance (8 cores/16 threads, 32 GB memory, gp3 volume/16000 IOPS/125 MB/s throughput):
+
+| **Processes** | **Seconds** |
+|:-------------:|:-----------:|
+|       1       |     N/A     |
+|       8       |     631     |
+|      16       |     491     |
+
+### Observations
+
+Ok I'm done with the dolma C4 baselines. on a c6a.4xlarge instance (using 8 processes), these are the timing I get:
+
+- Tagging with language id (fasttext) and C4 heuristics: 655s total, or 52s per jsonl per core
+- Filtering data: 631s total, or 50s per jsonl per core
+- Total time: 1286s, or 102s per jsonl per core. Projected CPU hours: (1286s * 8 cores / 23.92GB * 1024) / 3600 = 122 CPU Hours per TB.
+
+A couple of caveats:
+
+- I had to reformat the data to conform to dolma's format (create an ID field, rename "content" to "text"); I also gzip'ed files (dolma taggers can work with either regular or gzip'ed, although mixer expects gzip)
+- I am not using the C4 first, lang id after heuristics you all are using bc it's not automatically supported on dolma (you'd have to do it manually)
+- C4 rules implementation can still be implemented a bit. so far, the naughty words lookup step is the most expensive. Maybe a 15/20% speedup is possible.
+- The mixer requires fully downloading each file before it can be filtered. I could improve the rust code to allow streaming (taggers already stream)
diff --git a/examples/c4-like/mixer.yaml b/examples/c4-like/mixer.yaml
new file mode 100644
index 00000000..908b8148
--- /dev/null
+++ b/examples/c4-like/mixer.yaml
@@ -0,0 +1,48 @@
+streams:
+  - name: dcnlp_beta_ai2
+
+    documents:
+      - s3://ai2-llm/experimental/dcnlp_beta_ai2/v0/documents/*.jsonl.gz
+
+    output:
+      path: s3://ai2-llm/experimental/dcnlp_beta_ai2/v1/documents
+
+      # this should result in files around ~100MB. We keep this number small to
+      # ensure adequate parallelism during benchmarking.
+      max_size_in_bytes: 100_000_000
+      discard_fields:
+        - attributes
+        - metadata
+
+    attributes:
+      - c4_v2
+      - ft_lang_id_en_doc_v2
+    filter:
+      exclude:
+        # same filter as dolma: we remove documents whose English score is less than 0.5 according to fasttext
+        - $.attributes[?(@.ft_lang_id_en_doc_v2__ft_lang_id_en_doc_v2__en && @.ft_lang_id_en_doc_v2__ft_lang_id_en_doc_v2__en[0] && @.ft_lang_id_en_doc_v2__ft_lang_id_en_doc_v2__en[0][2] < 0.5)]
+
+        # exclude documents that contain a curly brace
+        - $.attributes[?(@.c4_v2__c4_v2__has_curly_brace && @.c4_v2__c4_v2__has_curly_brace[0] && @.c4_v2__c4_v2__has_curly_brace[0][2] > 0.5)]
+
+        # exclude documents that contain lorem ipsum
+        - $.attributes[?(@.c4_v2__c4_v2__has_lorem_ipsum && @.c4_v2__c4_v2__has_lorem_ipsum[0] && @.c4_v2__c4_v2__has_lorem_ipsum[0][2] > 0.5)]
+
+        # exclude documents that contain javascript
+        - $.attributes[?(@.c4_v2__c4_v2__has_javascript && @.c4_v2__c4_v2__has_javascript[0] && @.c4_v2__c4_v2__has_javascript[0][2] > 0.5)]
+
+        # exclude documents that contain one or more naughty words
+        - $.attributes[?(@.c4_v2__c4_v2__has_naughty_word && @.c4_v2__c4_v2__has_naughty_word[0] && @.c4_v2__c4_v2__has_naughty_word[0][2] > 0.5)]
+
+    span_replacement:
+      # remove lines that do not end in punctuation
+      - span: $.attributes.c4_v2__c4_v2__lines_with_no_ending_punctuation
+        min_score: 0.5
+        replacement: ""
+
+      # remove lines that are too short (less than 3 words as defined by C4 rules)
+      - span: $.attributes.c4_v2__c4_v2__lines_with_too_few_words
+        min_score: 0.5
+        replacement: ""
+
+processes: 8
diff --git a/examples/c4-like/reformat_trafilatura.py b/examples/c4-like/reformat_trafilatura.py
new file mode 100644
index 00000000..fb7b7662
--- /dev/null
+++ b/examples/c4-like/reformat_trafilatura.py
@@ -0,0 +1,86 @@
+import hashlib
+import json
+import sys
+from dataclasses import dataclass, field
+from queue import Queue
+from tempfile import TemporaryDirectory
+from typing import Any, Dict, Tuple, Union, cast
+
+import smart_open
+from omegaconf import MISSING
+from omegaconf.omegaconf import OmegaConf as om
+
+from dolma.core.parallel import BaseParallelProcessor
+
+
+class TrafilaturaReformatter(BaseParallelProcessor):
+    @classmethod
+    def increment_progressbar(  # type: ignore
+        cls,
+        queue: Queue[Union[Tuple[int, ...], None]],
+        /,
+        files: int = 0,
+        documents: int = 0,
+    ) -> Dict[str, int]:
+        return super().increment_progressbar(queue, files=files, documents=documents)
+
+    @classmethod
+    def process_single(
+        cls, source_path: str, destination_path: str, queue: Queue[Union[Tuple[int, ...], None]], **kwargs: Any
+    ):
+        documents = 0
+        interval = 10_000
+
+        destination_prefix = kwargs.get("destination_prefix", None)
+
+        new_destination_suffix = (
+            destination_path.replace(destination_prefix, "").lstrip("/").replace("/", "_") + ".gz"
+        )
+        destination_path = f"{destination_prefix.rstrip()}/{new_destination_suffix}"
+
+        with smart_open.open(source_path, "rt") as source_file, smart_open.open(
+            destination_path, "wt"
+        ) as destination_file:
+            for line in source_file:
+                document = json.loads(line)
+                documents += 1
+
+                transformed = {
+                    # use hash of the whole document as the id
+                    "id": hashlib.md5(line.encode("utf-8")).hexdigest(),
+                    "text": document["content"],
+                    "source": "trafilatura",
+                    "metadata": document,
+                }
+
+                if documents % interval == 0:
+                    cls.increment_progressbar(queue, documents=interval)
+
+                destination_file.write(json.dumps(transformed) + "\n")
+
+        cls.increment_progressbar(queue, files=1, documents=documents % interval)
+
+
+@dataclass
+class Config:
+    src: str = field(default_factory=lambda: MISSING)
+    dst: str = field(default_factory=lambda: MISSING)
+    proc: int = field(default=1)
+    debug: bool = field(default=False)
+
+
+def reformat_files(config: Config):
+    with TemporaryDirectory() as tempdir:
+        processor = TrafilaturaReformatter(
+            source_prefix=config.src,
+            destination_prefix=config.dst,
+            metadata_prefix=tempdir,
+            num_processes=config.proc,
+            debug=config.debug,
+        )
+        processor(destination_prefix=config.dst)
+
+
+if __name__ == "__main__":
+    config = om.merge(om.structured(Config), om.from_cli(sys.argv[1:]))
+    reformat_files(cast(Config, config))
diff --git a/examples/c4-like/tagger.yaml b/examples/c4-like/tagger.yaml
new file mode 100644
index 00000000..234b2006
--- /dev/null
+++ b/examples/c4-like/tagger.yaml
@@ -0,0 +1,8 @@
+documents:
+  - s3://ai2-llm/experimental/dcnlp_beta_ai2/v0/documents/*.gz
+
+taggers:
+  - c4_v2
+  - ft_lang_id_en_doc_v2
+
+processes: 8
diff --git a/pyproject.toml b/pyproject.toml
index d78c1598..4b3de0d7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "dolma"
-version = "0.8.0"
+version = "0.9.0"
 description = "Data filters"
 license = {text = "Apache-2.0"}
 readme = "README.md"
diff --git a/python/dolma/cli/analyzer.py b/python/dolma/cli/analyzer.py
index 9644aa3f..20f130e4 100644
--- a/python/dolma/cli/analyzer.py
+++ b/python/dolma/cli/analyzer.py
@@ -59,7 +59,7 @@ def run(cls, parsed_config: AnalyzerConfig):
             current_matching_documents = sum(1 for _ in glob_path(document))
             if current_matching_documents == 0:
                 # only raise a warning if no documents are found for a single path
-                logger.warn(f"No documents found for path {document}")
+                logger.warning("No documents found for path %s", document)
             total_matching_documents += current_matching_documents
 
         if total_matching_documents == 0:
diff --git a/python/dolma/cli/deduper.py b/python/dolma/cli/deduper.py
index c2e1eaae..3f09efb2 100644
--- a/python/dolma/cli/deduper.py
+++ b/python/dolma/cli/deduper.py
@@ -70,6 +70,10 @@ class DeduperConfig:
     processes: int = field(
         default=1, help="Number of processes to use for deduplication. If 1, no multiprocessing will be used."
     )
+    dryrun: bool = field(
+        default=False,
+        help="If true, only print the configuration and exit without running the deduper.",
+    )
 
 
 class DeduperCli(BaseCli):
@@ -103,7 +107,7 @@ def run(cls, parsed_config: DeduperConfig):
                 current_matching_documents = sum(1 for _ in glob_path(document))
                 if current_matching_documents == 0:
                     # only raise a warning if no documents are found for a single path
-                    logger.warn(f"No documents found for path {document}")
+                    logger.warning("No documents found for path %s", document)
                 total_matching_documents += current_matching_documents
 
             if total_matching_documents == 0:
@@ -137,4 +141,8 @@ def run(cls, parsed_config: DeduperConfig):
                 raise ValueError("At least one document must be specified")
 
             print_config(dict_config)
+            if parsed_config.dryrun:
+                logger.info("Exiting due to dryrun.")
+                return
+
             return deduper(dict_config)
diff --git a/python/dolma/cli/mixer.py b/python/dolma/cli/mixer.py
index 2d76423c..e92608eb 100644
--- a/python/dolma/cli/mixer.py
+++ b/python/dolma/cli/mixer.py
@@ -50,6 +50,10 @@ class MixerConfig:
     streams: List[StreamConfig] = field(default=[], help="List configurations of streams to be mixed")
     work_dir: WorkDirConfig = field(default=WorkDirConfig(), help="Configuration for temporary work directories.")
     processes: int = field(default=1, help="Number of processes to use for mixing. By default 1 process is used.")
+    dryrun: bool = field(
+        default=False,
+        help="If true, only print the configuration and exit without running the mixer.",
+    )
 
 
 class MixerCli(BaseCli):
@@ -100,7 +104,7 @@ def run(cls, parsed_config: MixerConfig):
                     current_matching_documents = sum(1 for _ in glob_path(document))
                     if current_matching_documents == 0:
                         # only raise a warning if no documents are found for a single path
-                        logger.warn(f"No documents found for path {document}")
+                        logger.warning("No documents found for path %s", document)
                     total_matching_documents += current_matching_documents
 
                 if total_matching_documents == 0:
@@ -128,4 +132,8 @@ def run(cls, parsed_config: MixerConfig):
                 raise DolmaConfigError("No streams to mix")
 
             print_config(dict_config)
-            return mixer(dict_config)
+            if parsed_config.dryrun:
+                logger.info("Exiting due to dryrun.")
+                return
+
+            mixer(dict_config)
diff --git a/python/dolma/cli/tagger.py b/python/dolma/cli/tagger.py
index 144ac69e..7e14d2a2 100644
--- a/python/dolma/cli/tagger.py
+++ b/python/dolma/cli/tagger.py
@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from pstats import SortKey
 from typing import List, Optional
 
 from rich.console import Console
@@ -13,6 +14,31 @@
 from dolma.core.runtime import create_and_run_tagger
 
 
+@dataclass
+class ProfilerConfig:
+    enable: bool = field(
+        default=False,
+        help="Whether to enable profiling.",
+    )
+    output: Optional[str] = field(
+        default=None,
+        help="Path to save the profiling output; if not provided, the output will be printed to stdout.",
+    )
+    steps: Optional[int] = field(
+        default=None,
+        help="List of steps to profile; if not provided, all steps will be profiled.",
+    )
+    sort_key: str = field(
+        default="tottime",
+        choices=[str(k) for k in SortKey._value2member_map_],  # pylint: disable=no-member,protected-access
+        help="Sort key for the profiling output.",
+    )
+    lines: int = field(
+        default=100,
+        help="Number of rows to print in the profiling output.",
+    )
+
+
 @dataclass
 class TaggerConfig:
     documents: List[str] = field(
@@ -47,7 +73,15 @@ class TaggerConfig:
         default=False,
         help="Whether to run in debug mode.",
     )
+    profile: ProfilerConfig = field(
+        default=ProfilerConfig(),
+        help="Whether to run in profiling mode.",
+    )
     work_dir: WorkDirConfig = field(default=WorkDirConfig(), help="Configuration for temporary work directories.")
+    dryrun: bool = field(
+        default=False,
+        help="If true, only print the configuration and exit without running the taggers.",
+    )
 
 
 class TaggerCli(BaseCli):
@@ -71,7 +105,7 @@ def run(cls, parsed_config: TaggerConfig):
                 current_matching_documents = sum(1 for _ in glob_path(document))
                 if current_matching_documents == 0:
                     # only raise a warning if no documents are found for a single path
-                    logger.warn(f"No documents found for path {document}")
+                    logger.warning("No documents found for path %s", document)
                 total_matching_documents += current_matching_documents
 
             if total_matching_documents == 0:
@@ -79,6 +113,10 @@ def run(cls, parsed_config: TaggerConfig):
                 raise DolmaConfigError(f"No documents found for paths {documents}.")
 
             print_config(parsed_config)
+            if parsed_config.dryrun:
+                logger.info("Exiting due to dryrun.")
+                return
+
             create_and_run_tagger(
                 documents=documents,
                 destination=parsed_config.destination,
@@ -88,6 +126,10 @@ def run(cls, parsed_config: TaggerConfig):
                 num_processes=parsed_config.processes,
                 experiment=parsed_config.experiment,
                 debug=parsed_config.debug,
+                profile_enable=parsed_config.profile.enable,
+                profile_output=parsed_config.profile.output,
+                profile_steps=parsed_config.profile.steps,
+                profile_sort_key=parsed_config.profile.sort_key,
             )
 
 
diff --git a/python/dolma/core/paths.py b/python/dolma/core/paths.py
index 2a618222..2a7153b8 100644
--- a/python/dolma/core/paths.py
+++ b/python/dolma/core/paths.py
@@ -94,6 +94,21 @@ def _unpathify(protocol: str, path: Path) -> str:
     return path_str
 
 
+def delete_file(path: str, ignore_missing: bool = False) -> bool:
+    """Delete a file."""
+
+    fs = _get_fs(path)
+    try:
+        fs.rm(path)
+        deleted = True
+    except FileNotFoundError as ex:
+        if not ignore_missing:
+            raise ex
+        deleted = False
+
+    return deleted
+
+
 def partition_path(path: str) -> Tuple[str, Tuple[str, ...], Tuple[str, ...]]:
     """Partition a path into its protocol, symbols before a glob, and symbols after a glob."""
     # split the path into its protocol and path components
diff --git a/python/dolma/core/runtime.py b/python/dolma/core/runtime.py
index 3f8382d2..a57a3fe4 100644
--- a/python/dolma/core/runtime.py
+++ b/python/dolma/core/runtime.py
@@ -1,3 +1,4 @@
+import io
 import logging
 import multiprocessing
 import tempfile
@@ -246,6 +247,9 @@ def process_single(
         # skip on failure
         skip_on_failure = kwargs.get("skip_on_failure", False)
 
+        # maximum numbers of lines to process
+        steps: Union[int, None] = kwargs.get("steps", None)
+
         # interval at which to update the progress bar; will double if it gets
         # too full
         update_interval = 1
@@ -276,6 +280,10 @@ def process_single(
                     # increment the number of documents processed so far
                     docs_cnt += 1
 
+                    if steps is not None and docs_cnt >= steps:
+                        # if we have reached the maximum number of steps, we break
+                        break
+
                     if docs_cnt % update_interval == 0:
                         # update the progress bar every 1000 documents to prevent
                         # buffering
@@ -286,22 +294,42 @@ def process_single(
                             # double the update interval if the queue is full
                             update_interval *= 2
 
-            except Exception as e:
+            except Exception as exp:
                 # handle any exception that might have occurred
-                msg = f"Failed to process {source_path} due to {e.__class__.__name__}: {' '.join(e.args)}"
-                if e.__class__.__name__ == "IncompleteReadError":
+                msg = f"Failed to process {source_path} due to {exp.__class__.__name__}: {' '.join(exp.args)}"
+                if exp.__class__.__name__ == "IncompleteReadError":
                     # Intermittent error that occurs when reading from S3
-                    raise DolmaRetryableFailure(msg) from e
+                    raise DolmaRetryableFailure(msg) from exp
                 else:
                     if skip_on_failure:
-                        raise DolmaShardError(msg) from e
+                        raise DolmaShardError(msg) from exp
                     else:
-                        raise DolmaFatalError(msg) from e
+                        raise DolmaFatalError(msg) from exp
 
         # increment the files progress bar
         cls.increment_progressbar(queue, files=1, documents=docs_cnt)
 
 
+@contextmanager
+def profiler(
+    output: Optional[str] = None,
+    sort_key: str = "tottime",
+    lines: int = 100,
+) -> Generator[None, None, None]:
+    import cProfile
+    import pstats
+
+    profile = cProfile.Profile()
+    profile.enable()
+    yield
+    profile.disable()
+
+    with ExitStack() as stack:
+        output_stream = io.StringIO() if output is None else stack.enter_context(smart_open.open(output, "w"))
+        ps = pstats.Stats(profile, stream=output_stream).sort_stats(sort_key)
+        ps.print_stats(lines)
+
+
 def create_and_run_tagger(
     documents: List[str],
     taggers: List[str],
@@ -314,6 +342,11 @@ def create_and_run_tagger(
     skip_on_failure: bool = False,
     retries_on_error: int = 0,
     num_processes: int = 1,
+    profile_enable: bool = False,
+    profile_output: Optional[str] = None,
+    profile_steps: Optional[int] = None,
+    profile_sort_key: str = "tottime",
+    profile_lines: int = 100,
 ):
     """This function creates a tagger and runs it on a list of documents.
 
@@ -339,6 +372,12 @@ def create_and_run_tagger(
         retries_on_error (int, optional): Number of times to retry processing a document if it fails.
             Defaults to 0 (fail immediately)
         num_processes (int, optional): Number of processes to use. Defaults to 1.
+        profile_enable (bool, optional): Whether to enable profiling. Defaults to False.
+        profile_output (Optional[str], optional): Path to save the profiling output; if not provided, the
+            output will be printed to stdout. Defaults to None.
+        profile_steps (Optional[int], optional): Number of steps to profile; if not provided, all steps will
+            be profiled. Defaults to None.
+        profile_sort_key (str, optional): Sort key for the profiling output. Defaults to 'tottime'.
     """
 
     # use placeholder experiment name if none is provided; raise an error if the placeholder name is used
@@ -350,33 +389,42 @@ def create_and_run_tagger(
     if destination is None:
         try:
             destination = _make_paths_from_substitution(documents, "documents", f"attributes/{experiment}")
-        except Exception as e:
-            raise RuntimeError("Could not make destination paths from documents paths") from e
+        except Exception as exp:
+            raise RuntimeError("Could not make destination paths from documents paths") from exp
     elif isinstance(destination, str):
         try:
             destination = _make_paths_from_prefix(documents, join_path(None, destination, experiment))
-        except Exception as e:
-            raise RuntimeError(f"Could not make destination paths from prefix {destination}") from e
+        except Exception as exp:
+            raise RuntimeError(f"Could not make destination paths from prefix {destination}") from exp
 
     metadata = metadata or tempfile.mkdtemp()
     if isinstance(metadata, str):
         try:
             metadata = _make_paths_from_prefix(documents, metadata)
-        except Exception as e:
-            raise RuntimeError(f"Could not make metadata paths from prefix {metadata}") from e
+        except Exception as exp:
+            raise RuntimeError(f"Could not make metadata paths from prefix {metadata}") from exp
 
         tagger = TaggerProcessor(
             source_prefix=documents,
             destination_prefix=destination,
             metadata_prefix=metadata,
-            debug=debug,
+            debug=debug or profile_enable,  # if profile is true, debug must be true
             seed=seed,
             ignore_existing=ignore_existing,
             retries_on_error=retries_on_error,
             num_processes=num_processes,
         )
-        tagger(
-            experiment_name=experiment,
-            taggers_names=taggers,
-            skip_on_failure=skip_on_failure,
-        )
+
+        with ExitStack() as stack:
+            if profile_enable:
+                # start profiling
+                stack.enter_context(
+                    profiler(output=profile_output, sort_key=profile_sort_key, lines=profile_lines)
+                )
+
+            tagger(
+                experiment_name=experiment,
+                taggers_names=taggers,
+                skip_on_failure=skip_on_failure,
+                steps=profile_steps,
+            )
diff --git a/python/dolma/taggers/c4.py b/python/dolma/taggers/c4.py
index b8f484ab..c5e6ae1e 100644
--- a/python/dolma/taggers/c4.py
+++ b/python/dolma/taggers/c4.py
@@ -1,6 +1,6 @@
 import logging
-import os
 from dataclasses import dataclass
+from pathlib import Path
 from typing import List, Set
 
 from ..core.data_types import DocResult, Document, Span
@@ -8,13 +8,10 @@
 from ..core.taggers import BaseTagger
 
 MIN_WORDS_PER_LINE = 3
-naughty_lines = (
-    open(os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "naughty_words_en.txt"))
-    .read()
-    .splitlines()
-)
-NAUGHTY_WORDS: Set[str] = set(w for w in naughty_lines if " " not in w)
-NAUGHTY_PHRASES: Set[str] = set(w for w in naughty_lines if " " in w)
+NAUGHTY_LINES = (Path(__file__).parent / "../data/naughty_words_en.txt").absolute().open().read().splitlines()
+NAUGHTY_WORDS: Set[str] = set(w for w in NAUGHTY_LINES if " " not in w)
+NAUGHTY_PHRASES: Set[str] = set(w for w in NAUGHTY_LINES if " " in w)
+EOL_PUNCTUATION = {".", "?", "!", '"'}
 
 
 @dataclass
@@ -85,3 +82,46 @@ def predict(self, doc: Document) -> DocResult:
         attrs = get_attributes(doc.text)
         result = DocResult(doc=doc, spans=attrs.as_spans())
         return result
+
+
+@TaggerRegistry.add("c4_v2")
+class FasterC4Tagger(BaseTagger):
+    def predict(self, doc: Document) -> DocResult:
+        spans: List[Span] = []
+        text = doc.text.lower()
+
+        if "{" in text:
+            spans.append(Span(0, len(doc.text), type="has_curly_brace"))
+
+        if "lorem ipsum" in text:
+            spans.append(Span(0, len(doc.text), type="has_lorem_ipsum"))
+
+        if "javascript" in text:
+            spans.append(Span(0, len(doc.text), type="has_javascript"))
+
+        if any(word in NAUGHTY_WORDS for word in text.split()) or any(
+            phrase in text for phrase in NAUGHTY_PHRASES
+        ):
+            spans.append(Span(0, len(doc.text), type="has_naughty_word"))
+
+        start = count = 0
+        for sent in text.split("\n"):
+            end = start + len(sent)
+            if end != len(text):
+                # account for the newline
+                end += 1
+
+            # strip any trailing whitespace
+            sent = sent.strip()
+
+            if not sent.endswith((".", "?", "!", '"')):
+                spans.append(Span(start, end, type="lines_with_no_ending_punctuation"))
+
+            if len(sent.split()) < MIN_WORDS_PER_LINE:
+                spans.append(Span(start, end, type="lines_with_too_few_words"))
+
+            count += 1
+            start = end
+
+        spans.append(Span(0, len(doc.text), type="line_count", score=count))
+        return DocResult(doc=doc, spans=spans)
diff --git a/scripts/stats.py b/scripts/stats.py
index 93546115..1edcb95d 100644
--- a/scripts/stats.py
+++ b/scripts/stats.py
@@ -206,7 +206,7 @@ def all(cls) -> Generator[Tuple[str, Type["BaseStatsProcessor"]], None, None]:
 
 class BaseStatsProcessor(BaseParallelProcessor):
     @classmethod
-    def increment_progressbar(  # type: ignore[override]
+    def increment_progressbar(  # type: ignore
         cls,
         queue: Queue[Union[Tuple[int, ...], None]],
         /,
@@ -404,6 +404,52 @@ def process_single(
             destination_file.write(json.dumps(stats, indent=2))
 
 
+@Registry.add
+class just_cc_dedup(books):
+    documents: str = "s3://ai2-llm/pretraining-data/sources/common-crawl/v1-c4-cleaned/documents/cc_en_*/*.gz"
+    stats: str = "s3://ai2-llm/stats/olmo-mix/v1/cc/just_cc_dedup/**/*.gz"
+
+    @classmethod
+    def process_single(
+        cls, source_path: str, destination_path: str, queue: Queue[Union[Tuple[int, ...], None]], **kwargs: Any
+    ):
+        dedup = source_path.replace("/documents/", "/attributes/dedupe_paragraphs/")
+
+        attr_decoder = msgspec.json.Decoder(OutputSpec)
+        stats = {
+            "dedupe_paragraphs_count": 0,
+            "dedupe_paragraphs_length": 0,
+            "dedupe_paragraphs_matches": 0,
+        }
+        documents = 0
+        interval = 10_000
+
+        with ExitStack() as stack:
+            try:
+                dedup_file = stack.enter_context(smart_open.open(dedup, "rb"))
+            except Exception:
+                return
+
+            for ln in dedup_file:
+                attrs = attr_decoder.decode(ln).attributes
+
+                # Duplicates stats
+                dups = [p for p in attrs.get("bff_duplicate_paragraph_spans", []) if p[1] - p[0] > 0]
+                stats["dedupe_paragraphs_count"] += len(dups)
+                stats["dedupe_paragraphs_length"] += sum(s[1] - s[0] for s in dups)
+                stats["dedupe_paragraphs_matches"] += 1 if dups else 0
+
+                documents += 1
+
+                if documents % interval == 0:
+                    cls.increment_progressbar(queue, documents=interval)
+
+        cls.increment_progressbar(queue, files=1, documents=documents % interval)
+
+        with smart_open.open(destination_path, "wt") as destination_file:
+            destination_file.write(json.dumps(stats, indent=2))
+
+
 @Registry.add
 class cc_v1_c4_cleaned(books):
     documents: str = "s3://ai2-llm/pretraining-data/sources/common-crawl/v1-c4-cleaned/documents/cc_en_*/*.gz"
@@ -518,6 +564,7 @@ def process_single(
             source_path.replace("/documents/", "/attributes/decontamination/"),
             source_path.replace("/documents/", "/attributes/hatespeech_nsfw_cc_v3/"),
             source_path.replace("/documents/", "/attributes/pii_detection/"),
+            source_path.replace("/documents/", "/attributes/dedupe_paragraphs/"),
         ]
 
         doc_decoder = msgspec.json.Decoder(InputSpec)
@@ -531,12 +578,15 @@ def process_single(
             "decontamination_count": 0,
             "decontamination_length": 0,
             "decontamination_matches": 0,
+            "dedupe_paragraphs_count": 0,
+            "dedupe_paragraphs_length": 0,
+            "dedupe_paragraphs_matches": 0,
             "hatespeech_nsfw_count": 0,
             "hatespeech_nsfw_length": 0,
             "hatespeech_nsfw_matches": 0,
             "pii_count": 0,
             "pii_length": 0,
-            "pii_matches": 0,
+            "pii_matches_le_5": 0,
             "pii_matches_gt_5": 0,
         }
         documents = 0
@@ -544,7 +594,11 @@ def process_single(
 
         with ExitStack() as stack:
             doc_file = stack.enter_context(smart_open.open(source_path, "rb"))
-            atts_files = [stack.enter_context(smart_open.open(path, "rb")) for path in attributes]
+
+            try:
+                atts_files = [stack.enter_context(smart_open.open(path, "rb")) for path in attributes]
+            except Exception:
+                return
 
             for doc_line, *attr_lines in zip(doc_file, *atts_files):
                 doc = doc_decoder.decode(doc_line)
@@ -555,12 +609,6 @@ def process_single(
                 for line in attr_lines:
                     attrs.update(attr_decoder.decode(line).attributes)
 
-                # # C4 stats
-                # c4_removal = attrs.get('c4_rules__c4_v1__lines_with_no_ending_punctuation', [])
-                # stats["c4_count"] += len(c4_removal)
-                # stats["c4_length"] += sum(s[-1] for s in c4_removal)
-                # stats["c4_matches"] += 1 if c4_removal else 0
-
                 # Gopher stats
                 gopher_removal = cls.gopher_rules(attrs)
                 stats["gopher_count"] += len(gopher_removal)
@@ -599,9 +647,15 @@ def process_single(
                 )
                 stats["pii_count"] += len(pii_removal)
                 stats["pii_length"] += sum(s[1] - s[0] for s in pii_removal)
-                stats["pii_matches"] += 1 if pii_removal else 0
+                stats["pii_matches_le_5"] += 1 if 0 < len(pii_removal) <= 5 else 0
                 stats["pii_matches_gt_5"] += 1 if len(pii_removal) > 5 else 0
 
+                # Duplicates stats
+                dups = [p for p in attrs.get("bff_duplicate_paragraph_spans", []) if p[1] - p[0] > 0]
+                stats["dedupe_paragraphs_count"] += len(dups)
+                stats["dedupe_paragraphs_length"] += sum(s[1] - s[0] for s in dups)
+                stats["dedupe_paragraphs_matches"] += 1 if dups else 0
+
                 documents += 1
 
                 if documents % interval == 0:
@@ -755,56 +809,77 @@ class StackInputSpec(InputSpec):
 
 
 @Registry.add
-class stack(BaseStatsProcessor):
+class stack_v2(books):
+    documents: str = "s3://ai2-llm/pretraining-data/sources/stack-dedup/v1-mixer/documents/*.gz"
+    stats: str = "s3://ai2-llm/stats/olmo-mix/v1/stack/v1-mixer/*.gz"
+
     @classmethod
     def process_single(
         cls, source_path: str, destination_path: str, queue: Queue[Union[Tuple[int, ...], None]], **kwargs: Any
     ):
         attrs_basic = source_path.replace("/documents/", "/attributes/basic/")
-        attrs_code_secrets = source_path.replace("/documents/", "/attributes/code_secrets/")
+        attrs_code_secrets = source_path.replace("/documents/", "/attributes/rpj-heuristics/")
         # attrs_dedupe_documents = source_path.replace("/documents/", "/attributes/dedupe_documents/")
-        attrs_pii = source_path.replace("/documents/", "/attributes/pii/")
+        # attrs_pii = source_path.replace("/documents/", "/attributes/pii/")
 
         documents_decoder = msgspec.json.Decoder(StackInputSpec)
         attributes_decoder = msgspec.json.Decoder(OutputSpec)
 
         interval = 10_000
 
-        counts: dict = {"extension": defaultdict(int), "license": defaultdict(int), "documents": 0, "tokens": 0}
+        counts: dict = {
+            "extension": defaultdict(int),
+            "license": defaultdict(int),
+            "length": 0,
+            "count": 0,
+            "rpj_count": 0,
+            "rpj_length": 0,
+            "rpj_matches": 0,
+        }
         cnt = 0
 
         with ExitStack() as stack:
             doc_file = stack.enter_context(smart_open.open(source_path, "rb"))
-            attrs_basic_file = stack.enter_context(smart_open.open(attrs_basic, "rb"))
-            attrs_code_secrets_file = stack.enter_context(smart_open.open(attrs_code_secrets, "rb"))
-            # attrs_dedupe_documents_file = stack.enter_context(smart_open.open(attrs_dedupe_documents, "rb"))
-            attrs_pii_file = stack.enter_context(smart_open.open(attrs_pii, "rb"))
+            attributes_files = [
+                stack.enter_context(smart_open.open(attrs_basic, "rb")),
+                stack.enter_context(smart_open.open(attrs_code_secrets, "rb")),
+                # stack.enter_context(smart_open.open(attrs_dedupe_documents, "rb")),
+                # stack.enter_context(smart_open.open(attrs_pii, "rb")),
+            ]
 
             # with smart_open.open(source_path, "rb") as doc_file, smart_open.open(attrs_path, "rb") as attrs_file:
-            for source_line, *attributes_line in zip(
-                doc_file,
-                attrs_basic_file,
-                attrs_code_secrets_file,
-                # attrs_dedupe_documents_file,
-                attrs_pii_file,
-            ):
+            for source_line, *attributes_line in zip(doc_file, *attributes_files):
                 cnt += 1
 
-                document = documents_decoder.decode(source_line)
+                doc = documents_decoder.decode(source_line)
                 for ln in attributes_line:
                     attributes = attributes_decoder.decode(ln)
-                    document.attributes.update(attributes.attributes)
+                    doc.attributes.update(attributes.attributes)
 
-                if document.attributes["basic__random_number_v1__random"][-1][-1] > 0.996:
+                if doc.attributes["basic__random_number_v1__random"][-1][-1] > 0.996:
                     # test set; see
                     # https://github.com/allenai/LLM/blob/642d0fad3fb2efd816af507250c4c65c8678cb44/pretrain_data/the_stack/v2-mixer/ablations/v2-mixer-held-out.json#L15
                     continue
 
-                counts["documents"] += 1
-                counts["tokens"] += len(blingfire.text_to_words(document.text).split())
-
-                counts["extension"][document.metadata["ext"]] += 1
-                for license in document.metadata["max_forks_repo_licenses"]:
+                spans = []
+                if doc.attributes["rpj_heuristics__code_redpajama_taggers_v1__max_line_length_doc"][0][2] > 1000:
+                    spans.append((0, len(doc.text), 1.0))
+                elif doc.attributes["rpj_heuristics__code_redpajama_taggers_v1__avg_line_length_doc"][0][2] > 100:
+                    spans.append((0, len(doc.text), 1.0))
+                elif doc.attributes["rpj_heuristics__code_redpajama_taggers_v1__alnum_prop_doc"][0][2] < 0.25:
+                    spans.append((0, len(doc.text), 1.0))
+                elif doc.attributes["rpj_heuristics__code_redpajama_taggers_v1__alpha_token_prop_doc"][0][2] < 1.5:
+                    spans.append((0, len(doc.text), 1.0))
+
+                counts["rpj_count"] += len(spans)
+                counts["rpj_length"] += sum(s[1] - s[0] for s in spans)
+                counts["rpj_matches"] += 1 if spans else 0
+
+                counts["count"] += 1
+                counts["length"] += len(doc.text)
+
+                counts["extension"][doc.metadata["ext"]] += 1
+                for license in doc.metadata["max_forks_repo_licenses"]:
                     counts["license"][license] += 1
 
                 cnt += 1
@@ -817,50 +892,91 @@ def process_single(
         with smart_open.open(destination_path, "wt") as destination_file:
             destination_file.write(json.dumps(counts, indent=2))
 
-    @staticmethod
-    # read all paths in using threads
-    def _read_json(path: str) -> dict:
-        with smart_open.open(path, "rt") as source_file:
-            content = msgspec.json.decode(source_file.read())
-            return content
 
-    @classmethod
-    def _merge_dicts(cls, d1, d2):
-        d1 = copy.deepcopy(d1)
-        for k, v in d2.items():
-            if isinstance(v, dict):
-                d1[k] = cls._merge_dicts(d1.get(k, {}), v)
-            else:
-                d1[k] = d1.get(k, 0) + v
-        return d1
+@Registry.add
+class stack_v3(stack_v2):
+    documents: str = "s3://ai2-llm/pretraining-data/sources/stack-dedup/v2-mixer/documents/*.gz"
+    stats: str = "s3://ai2-llm/stats/olmo-mix/v1/stack/v2-mixer/*.gz"
 
     @classmethod
-    def cli(cls, num_workers: int = 1, debug: bool = False, **process_single_kwargs: Any) -> None:
-        with TemporaryDirectory() as tempdir:
-            documents = "s3://ai2-llm/pretraining-data/sources/stack-dedup/v2-mixer/documents/*.gz"
-            stats = "s3://ai2-llm/stats/olmo-mix/v1/code/stack"
-            metadata = os.path.join(tempdir, "stack")
+    def process_single(
+        cls, source_path: str, destination_path: str, queue: Queue[Union[Tuple[int, ...], None]], **kwargs: Any
+    ):
+        attrs_basic = source_path.replace("/documents/", "/attributes/basic/")
+        # attrs_code_secrets = source_path.replace("/documents/", "/attributes/rpj-heuristics/")
+        attrs_dedupe_documents = source_path.replace("/documents/", "/attributes/dedupe_documents/")
+        attrs_pii = source_path.replace("/documents/", "/attributes/pii/")
 
-            processor = cls(
-                source_prefix=documents,
-                destination_prefix=stats,
-                metadata_prefix=metadata,
-                num_processes=num_workers,
-                debug=debug,
-            )
-            processor(**process_single_kwargs)
+        documents_decoder = msgspec.json.Decoder(StackInputSpec)
+        attributes_decoder = msgspec.json.Decoder(OutputSpec)
 
-        paths = list(glob_path(f"{stats}/*.gz"))
-        counts: dict = {}
+        interval = 10_000
 
-        with multiprocessing.Pool(num_workers) as pool:
-            data = (cls._read_json(path) for path in paths) if debug else pool.imap(cls._read_json, paths)
+        counts: dict = {
+            "extension": defaultdict(int),
+            "license": defaultdict(int),
+            "length": 0,
+            "count": 0,
+            "rpj_count": 0,
+            "rpj_length": 0,
+            "rpj_matches": 0,
+        }
+        cnt = 0
 
-            for content in tqdm.tqdm(data, desc="Merging code stats", unit=" files", total=len(paths)):
-                counts = cls._merge_dicts(counts, content)
+        with ExitStack() as stack:
+            doc_file = stack.enter_context(smart_open.open(source_path, "rb"))
+            attributes_files = [
+                stack.enter_context(smart_open.open(attrs_basic, "rb")),
+                stack.enter_context(smart_open.open(attrs_dedupe_documents, "rb")),
+                stack.enter_context(smart_open.open(attrs_pii, "rb")),
+            ]
 
-        with smart_open.open("s3://ai2-llm/stats/olmo-mix/v1/code/summary.json", "wt") as destination_file:
-            destination_file.write(json.dumps(counts, indent=2, sort_keys=True))
+            # with smart_open.open(source_path, "rb") as doc_file, smart_open.open(attrs_path, "rb") as attrs_file:
+            for source_line, *attributes_line in zip(doc_file, *attributes_files):
+                cnt += 1
+
+                doc = documents_decoder.decode(source_line)
+                for ln in attributes_line:
+                    attributes = attributes_decoder.decode(ln)
+                    doc.attributes.update(attributes.attributes)
+
+                if doc.attributes["basic__random_number_v1__random"][-1][-1] > 0.996:
+                    # test set; see
+                    # https://github.com/allenai/LLM/blob/642d0fad3fb2efd816af507250c4c65c8678cb44/pretrain_data/the_stack/v2-mixer/ablations/v2-mixer-held-out.json#L15
+                    continue
+
+                breakpoint()
+
+                spans = []
+                if doc.attributes["rpj_heuristics__code_redpajama_taggers_v1__max_line_length_doc"][0][2] > 1000:
+                    spans.append((0, len(doc.text), 1.0))
+                elif doc.attributes["rpj_heuristics__code_redpajama_taggers_v1__avg_line_length_doc"][0][2] > 100:
+                    spans.append((0, len(doc.text), 1.0))
+                elif doc.attributes["rpj_heuristics__code_redpajama_taggers_v1__alnum_prop_doc"][0][2] < 0.25:
+                    spans.append((0, len(doc.text), 1.0))
+                elif doc.attributes["rpj_heuristics__code_redpajama_taggers_v1__alpha_token_prop_doc"][0][2] < 1.5:
+                    spans.append((0, len(doc.text), 1.0))
+
+                counts["rpj_count"] += len(spans)
+                counts["rpj_length"] += sum(s[1] - s[0] for s in spans)
+                counts["rpj_matches"] += 1 if spans else 0
+
+                counts["count"] += 1
+                counts["length"] += len(doc.text)
+
+                counts["extension"][doc.metadata["ext"]] += 1
+                for license in doc.metadata["max_forks_repo_licenses"]:
+                    counts["license"][license] += 1
+
+                cnt += 1
+
+                if cnt % interval == 0:
+                    cls.increment_progressbar(queue, documents=interval)
+
+        cls.increment_progressbar(queue, files=1, documents=cnt % interval)
+
+        with smart_open.open(destination_path, "wt") as destination_file:
+            destination_file.write(json.dumps(counts, indent=2))
 
 
 # # # BELOW HERE: AGGREGATION # # #
diff --git a/src/s3_util.rs b/src/s3_util.rs
index 0a0a9fec..6e329813 100644
--- a/src/s3_util.rs
+++ b/src/s3_util.rs
@@ -154,6 +154,7 @@ pub fn find_objects_matching_patterns(
                 )
                 .unwrap()
             } else {
+                log::info!("Listing objects in bucket={}, prefix={}", bucket, key);
                 rt.block_on(
                     s3_client
                         .list_objects_v2()
@@ -164,24 +165,42 @@ pub fn find_objects_matching_patterns(
                 )
                 .unwrap()
             };
-            resp.contents().unwrap_or_default().iter().for_each(|obj| {
-                let s3_url = format!("s3://{}/{}", bucket, obj.key().unwrap());
-                stream_inputs.push(s3_url);
-            });
-            suffix.iter().for_each(|s| {
-                resp.common_prefixes()
-                    .unwrap_or_default()
-                    .iter()
-                    .for_each(|sub_folder| {
-                        let mut full_path = sub_folder.prefix().unwrap().to_owned();
-                        full_path.push_str(s);
-                        let s3_url = format!("s3://{}/{}", bucket, full_path);
-                        stream_inputs.push(s3_url);
+
+            // unrwrap resp.contents() and create a to_validate_stream_inputs vector
+            // containing all prefixes that do NOT end with "/"
+            let to_validate_stream_inputs: Vec<String> = resp
+                .contents()
+                .unwrap_or_default()
+                .iter()
+                .filter_map(|prefix| {
+                    if !prefix.key().unwrap().ends_with("/") {
+                        Some(format!("s3://{}/{}", bucket, prefix.key().unwrap()))
+                    } else {
+                        None
+                    }
+                })
+                .collect();
+
+            match suffix {
+                None => {
+                    // if suffix is none, push all the objects in to_validate_stream_inputs to stream_inputs
+                    to_validate_stream_inputs.iter().for_each(|path| {
+                        stream_inputs.push(path.to_owned());
                     });
-            });
+                }
+                _ => {
+                    // push only the objects that match the suffix to stream_inputs
+                    to_validate_stream_inputs.iter().for_each(|path| {
+                        if path.ends_with(suffix.clone().unwrap().as_str()) {
+                            stream_inputs.push(path.to_owned());
+                        }
+                    });
+                }
+            }
             token = resp.next_continuation_token().map(String::from);
             has_more = token.is_some();
         }
+
         log::info!(
             "Found {} objects for pattern \"{}\"",
             stream_inputs.len() - start_size,


Test patch:
diff --git a/tests/python/test_c4.py b/tests/python/test_c4.py
new file mode 100644
index 00000000..871ea730
--- /dev/null
+++ b/tests/python/test_c4.py
@@ -0,0 +1,123 @@
+"""
+
+Unit tests for taggers/*.py
+
+@kylel
+
+"""
+
+from unittest import TestCase
+
+from dolma.core.data_types import Document
+from dolma.taggers.c4 import C4Tagger, FasterC4Tagger
+
+
+class TestC4Tagger(TestCase):
+    def setUp(self):
+        self.tagger = C4Tagger()
+
+    def test_curly_braces_filter(self):
+        doc = Document(source="", version="", id="", text="This is a test.")
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual([r.to_json() for r in result.spans if r.type == "has_curly_brace"], [])
+
+        doc = Document(source="", version="", id="", text="This is a test {")
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual(
+            [r.to_json() for r in result.spans if r.type == "has_curly_brace"],
+            [{"start": 0, "end": 16, "type": "has_curly_brace", "score": 1.0}],
+        )
+
+    def test_javascript_filter(self):
+        doc = Document(source="", version="", id="", text="This is a test.")
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual([r.to_json() for r in result.spans if r.type == "has_javascript"], [])
+
+        doc = Document(source="", version="", id="", text="This is a test javascript")
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual(
+            [r.to_json() for r in result.spans if r.type == "has_javascript"],
+            [{"start": 0, "end": 25, "type": "has_javascript", "score": 1.0}],
+        )
+
+    def test_lorem_ipsum_filter(self):
+        doc = Document(source="", version="", id="", text="This is a test.")
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual([r.to_json() for r in result.spans if r.type == "has_lorem_ipsum"], [])
+
+        doc = Document(source="", version="", id="", text="This is a lorem ipsum test\nmore test.")
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual(
+            [r.to_json() for r in result.spans if r.type == "has_lorem_ipsum"],
+            # the end should be 37 because lorem ipsum is detected at the document level
+            [{"start": 0, "end": 37, "type": "has_lorem_ipsum", "score": 1.0}],
+        )
+
+    def test_line_ends_with_no_punctuation(self):
+        doc = Document(
+            source="",
+            version="",
+            id="",
+            text="This is a test.\nIt has more;\nA trailing space! \nShould be good on this one.\nThis one is bad\n",
+        )
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual(
+            [r.to_json() for r in result.spans if r.type == "lines_with_no_ending_punctuation"],
+            [
+                {"start": 16, "end": 29, "type": "lines_with_no_ending_punctuation", "score": 1.0},
+                {"start": 76, "end": 92, "type": "lines_with_no_ending_punctuation", "score": 1.0},
+                {"start": 92, "end": 92, "type": "lines_with_no_ending_punctuation", "score": 1.0},
+            ],
+        )
+
+    def test_lines_with_too_few_words(self):
+        doc = Document(
+            source="",
+            version="",
+            id="",
+            text="Short!\nThis short.\nBarely above the limit!\nthis is last frontier.",
+        )
+
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual(
+            [r.to_json() for r in result.spans if r.type == "lines_with_too_few_words"],
+            [
+                {"start": 0, "end": 7, "type": "lines_with_too_few_words", "score": 1.0},
+                {"start": 7, "end": 19, "type": "lines_with_too_few_words", "score": 1.0},
+            ],
+        )
+
+    def test_naughty_words(self):
+        doc = Document(source="", version="", id="", text="This sentence has no bad words.")
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual([r.to_json() for r in result.spans if r.type == "has_naughty_word"], [])
+
+        doc = Document(source="", version="", id="", text="This sentence mentions viagra as a bad word.")
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual(
+            [r.to_json() for r in result.spans if r.type == "has_naughty_word"],
+            [{"start": 0, "end": len(doc.text), "type": "has_naughty_word", "score": 1.0}],
+        )
+
+        doc = Document(
+            source="", version="", id="", text="This sentence has ass, but not a bad word because of comma."
+        )
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual([r.to_json() for r in result.spans if r.type == "has_naughty_word"], [])
+
+        doc = Document(
+            source="",
+            version="",
+            id="",
+            text="If I say strap on because we are on a plane, it's still a bad word.",
+        )
+        result = self.tagger.predict(doc=doc)
+        self.assertEqual(
+            [r.to_json() for r in result.spans if r.type == "has_naughty_word"],
+            [{"start": 0, "end": len(doc.text), "type": "has_naughty_word", "score": 1.0}],
+        )
+
+
+class TestFasterC4Tagger(TestC4Tagger):
+    def setUp(self):
+        self.tagger = FasterC4Tagger()
diff --git a/tests/python/test_taggers.py b/tests/python/test_gopher.py
similarity index 100%
rename from tests/python/test_taggers.py
rename to tests/python/test_gopher.py
diff --git a/tests/python/test_mixer.py b/tests/python/test_mixer.py
index d4b81bca..8f739a49 100644
--- a/tests/python/test_mixer.py
+++ b/tests/python/test_mixer.py
@@ -71,7 +71,7 @@ def test_local_input_remote_output(self):
         if self.remote_test_prefix is None:
             return self.skipTest("Skipping AWS tests")
 
-        with open(MIXER, "r") as f:
+        with open(MIXER, mode="r", encoding="utf8") as f:
             config = json.load(f)
 
         # keep track of local output path
@@ -97,7 +97,7 @@ def test_remote_input_remote_output(self):
         if self.remote_test_prefix is None:
             return self.skipTest("Skipping AWS tests")
 
-        with open(MIXER, "r") as f:
+        with open(MIXER, mode="r", encoding="utf8") as f:
             config = json.load(f)
 
         # keep track of local output path
@@ -127,7 +127,7 @@ def test_remote_input_local_output(self):
         if self.remote_test_prefix is None:
             return self.skipTest("Skipping AWS tests")
 
-        with open(MIXER, "r") as f:
+        with open(MIXER, mode="r", encoding="utf8") as f:
             config = json.load(f)
 
         # keep track of local output path