Content Processing & Cleaning

1
from nemo_curator.core.client import RayClient
2
from nemo_curator.pipeline import Pipeline
3
from nemo_curator.stages.text.io.reader import JsonlReader
4
from nemo_curator.stages.text.io.writer import JsonlWriter
5
from nemo_curator.stages.text.modifiers import Modify
6
from nemo_curator.stages.text.modifiers.string import UrlRemover, NewlineNormalizer
7
from nemo_curator.stages.text.modifiers.unicode import UnicodeReformatter
8
9
# Initialize Ray client
10
ray_client = RayClient()
11
ray_client.start()
12
13
# Create a comprehensive cleaning pipeline
14
processing_pipeline = Pipeline(
15
    name="content_processing_pipeline",
16
    description="Comprehensive text cleaning and processing"
17
)
18
19
# Load dataset
20
reader = JsonlReader(file_paths="input_data/")
21
processing_pipeline.add_stage(reader)
22
23
# Fix Unicode encoding issues
24
processing_pipeline.add_stage(
25
    Modify(modifier_fn=UnicodeReformatter(), input_fields="text")
26
)
27
28
# Standardize newlines
29
processing_pipeline.add_stage(
30
    Modify(modifier_fn=NewlineNormalizer(), input_fields="text")
31
)
32
33
# Remove URLs
34
processing_pipeline.add_stage(
35
    Modify(modifier_fn=UrlRemover(), input_fields="text")
36
)
37
38
# Save the processed dataset
39
writer = JsonlWriter(path="processed_output/")
40
processing_pipeline.add_stage(writer)
41
42
# Execute pipeline
43
results = processing_pipeline.run()
44
45
# Stop Ray client
46
ray_client.stop()