Common Crawl

1
from nemo_curator.core.client import RayClient
2
from nemo_curator.pipeline import Pipeline
3
from nemo_curator.stages.text.download import CommonCrawlDownloadExtractStage
4
from nemo_curator.stages.text.io.writer import JsonlWriter
5
6
def main():
7
    # Initialize Ray client
8
    ray_client = RayClient()
9
    ray_client.start()
10
11
    # Create pipeline
12
    pipeline = Pipeline(
13
        name="common_crawl_pipeline",
14
        description="Download and process Common Crawl data"
15
    )
16
17
    # Add Common Crawl processing stage
18
    cc_stage = CommonCrawlDownloadExtractStage(
19
        start_snapshot="2020-50",  # YYYY-WW format for CC-MAIN
20
        end_snapshot="2020-50",
21
        download_dir="./cc_downloads",
22
        crawl_type="main",  # or "news"
23
        use_aws_to_download=True,  # Faster S3 downloads (requires s5cmd)
24
        url_limit=10,  # Limit number of WARC files for testing
25
        record_limit=1000,  # Limit records per WARC file
26
    )
27
    pipeline.add_stage(cc_stage)
28
29
    # Add output writer stage
30
    writer = JsonlWriter("./cc_output")
31
    pipeline.add_stage(writer)
32
33
    # Run pipeline
34
    results = pipeline.run()
35
36
    # Stop Ray client
37
    ray_client.stop()
38
39
if __name__ == "__main__":
40
    main()

1
from nemo_curator.core.client import RayClient
2
from nemo_curator.pipeline import Pipeline
3
from nemo_curator.stages.text.download import CommonCrawlDownloadExtractStage
4
from nemo_curator.stages.text.io.writer import JsonlWriter
5
6
DOWNLOAD_DIR = "./cc_downloads"
7
OUTPUT_DIR = "./cc_output"
8
9
10
def main():
11
    ray_client = RayClient()
12
    ray_client.start()
13
14
    try:
15
        pipeline = Pipeline(
16
            name="common_crawl_resume_example",
17
            description="Download and process Common Crawl data with resumable downloads",
18
        )
19
        pipeline.add_stage(
20
            CommonCrawlDownloadExtractStage(
21
                start_snapshot="2020-50",
22
                end_snapshot="2020-50",
23
                download_dir=DOWNLOAD_DIR,
24
                crawl_type="main",
25
                use_aws_to_download=True,
26
            )
27
        )
28
        pipeline.add_stage(JsonlWriter(OUTPUT_DIR))
29
30
        # If this run is interrupted, run the same script again. Curator will
31
        # reuse completed files in DOWNLOAD_DIR and fetch only missing WARC files.
32
        pipeline.run()
33
    finally:
34
        ray_client.stop()
35
36
37
if __name__ == "__main__":
38
    main()

1
from nemo_curator.stages.text.io.writer import ParquetWriter
2
3
# Replace the JSONL writer with ParquetWriter
4
writer = ParquetWriter("./cc_output_parquet")
5
pipeline.add_stage(writer)

1
from nemo_curator.stages.text.download.html_extractors import ResiliparseExtractor
2
from nemo_curator.stages.text.download.html_extractors import TrafilaturaExtractor
3
4
# Use Resiliparse for extraction
5
cc_stage = CommonCrawlDownloadExtractStage(
6
    start_snapshot="2020-50",
7
    end_snapshot="2020-50",
8
    download_dir="./downloads",
9
    html_extraction=ResiliparseExtractor(
10
        required_stopword_density=0.25,
11
        main_content=True
12
    )
13
)
14
15
# Or use Trafilatura with custom parameters
16
cc_stage = CommonCrawlDownloadExtractStage(
17
    start_snapshot="2020-50", 
18
    end_snapshot="2020-50",
19
    download_dir="./downloads",
20
    html_extraction=TrafilaturaExtractor(
21
        min_extracted_size=200,
22
        max_repetitions=3
23
    )
24
)

1
# Define custom stop words for specific languages
2
stop_lists = {
3
    "ENGLISH": frozenset(["the", "and", "is", "in", "for", "where", "when", "to", "at"]),
4
    "SPANISH": frozenset(["el", "la", "de", "que", "y", "en", "un", "es", "se", "no"])
5
}
6
7
cc_stage = CommonCrawlDownloadExtractStage(
8
    start_snapshot="2020-50",
9
    end_snapshot="2020-50", 
10
    download_dir="./downloads",
11
    stop_lists=stop_lists
12
)

1
from nemo_curator.stages.text.download.common_crawl.download import CommonCrawlWARCReader
2
3
# HTTPS transport (default)
4
warc_reader = CommonCrawlWARCReader(
5
    warc_filename_col="warc_filename",
6
    warc_record_offset_col="warc_record_offset",
7
    warc_record_length_col="warc_record_length",
8
    max_workers=16,
9
)
10
11
# S3 transport
12
warc_reader = CommonCrawlWARCReader(
13
    warc_filename_col="warc_filename",
14
    warc_record_offset_col="warc_record_offset",
15
    warc_record_length_col="warc_record_length",
16
    use_s3=True,
17
    max_workers=16,
18
)

1
cc_stage = CommonCrawlDownloadExtractStage(
2
    start_snapshot="2020-08",  # YYYY-MM format for CC-NEWS
3
    end_snapshot="2020-10",
4
    download_dir="./news_downloads",
5
    crawl_type="news"  # Use CC-NEWS instead of CC-MAIN
6
)

1
cc_stage = CommonCrawlDownloadExtractStage(
2
    start_snapshot="2020-50",
3
    end_snapshot="2020-50",
4
    download_dir="/fast_storage/cc_downloads",
5
    use_aws_to_download=True,  # Faster S3 downloads
6
    verbose=False,  # Reduce logging overhead
7
    # Remove limits for full processing
8
    # url_limit=None,
9
    # record_limit=None
10
)

1
# Increase recycling frequency for very memory-constrained environments
2
cc_stage = CommonCrawlDownloadExtractStage(
3
    start_snapshot="2020-50",
4
    end_snapshot="2020-50",
5
    download_dir="./downloads",
6
    extractor_max_calls_per_worker=1,  # Recycle after every task
7
)
8
9
# Disable worker recycling (not recommended for large jobs with jusText)
10
cc_stage = CommonCrawlDownloadExtractStage(
11
    start_snapshot="2020-50",
12
    end_snapshot="2020-50",
13
    download_dir="./downloads",
14
    extractor_max_calls_per_worker=None,  # No recycling
15
)