Resumable Processing

This guide explains strategies to make large-scale data operations resumable.

Why Resumable Processing Matters

Large datasets can trigger interruptions due to:

System timeouts
Hardware failures
Network issues
Resource constraints
Scheduled maintenance

NeMo Curator provides built-in functionality for resuming operations from where they left off.

How it Works

The resumption approach works by:

Examining filenames in the input directory using get_all_file_paths_under()
Comparing them with filenames in the output directory
Identifying unprocessed files by comparing file counts or specific file lists
Rerunning the pipeline on remaining files

This approach works best when you:

Use consistent directory structures for input and output
Process files in batches using files_per_partition to manage memory usage
Create checkpoints by writing intermediate results to disk

Practical Patterns for Resumable Processing

1. Process remaining files using directory comparison

Use file listing utilities to identify unprocessed files and process them directly:

1 from nemo_curator.utils.file_utils import get_all_file_paths_under
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 
6 # Get all input files
7 input_files = get_all_file_paths_under(
8     "input_directory/", 
9     recurse_subdirectories=True, 
10     keep_extensions=[".jsonl"]
11 )
12 
13 # Get already processed output files
14 output_files = get_all_file_paths_under(
15     "output_directory/", 
16     recurse_subdirectories=True, 
17     keep_extensions=[".jsonl"]
18 )
19 
20 # Simple approach: if output directory has fewer files than input, 
21 # process all remaining inputs
22 if len(output_files) < len(input_files):
23     # Process remaining files
24     pipeline = Pipeline(name="resumable_processing")
25     
26     # Read input files
27     reader = JsonlReader(file_paths=input_files, fields=["text", "id"])
28     pipeline.add_stage(reader)
29     
30     # Add your processing stages here
31     # pipeline.add_stage(your_processing_stage)
32     
33     # Write results
34     writer = JsonlWriter(path="output_directory/")
35     pipeline.add_stage(writer)
36     
37     # Execute pipeline
38     pipeline.run()

2. Batch processing with file partitioning

Control memory usage and enable checkpoint creation by using NeMo Curator’s built-in file partitioning:

1 from nemo_curator.pipeline import Pipeline
2 from nemo_curator.stages.text.io.reader import JsonlReader
3 from nemo_curator.stages.text.io.writer import JsonlWriter
4 
5 # Process files in smaller batches using files_per_partition
6 pipeline = Pipeline(name="batch_processing")
7 
8 # JsonlReader automatically handles file partitioning
9 reader = JsonlReader(
10     file_paths="input_directory/", 
11     files_per_partition=64,  # Process 64 files at a time
12     fields=["text", "id"]
13 )
14 pipeline.add_stage(reader)
15 
16 # Add your processing stages here
17 # pipeline.add_stage(your_processing_stage)
18 
19 # Write results
20 writer = JsonlWriter(path="output_directory/")
21 pipeline.add_stage(writer)
22 
23 # Execute pipeline - processes files in batches automatically
24 pipeline.run()