Implementation:NVIDIA NeMo Curator FileUtils

Knowledge Sources	NVIDIA NeMo Curator
Domains	File I/O, Filesystem Operations, Cloud Storage, Data Curation
Last Updated	2026-02-14 00:00 GMT

Overview

Provides comprehensive filesystem operations for file discovery, directory management, output mode handling, and safe tar extraction, supporting both local and remote (cloud) filesystems via fsspec.

Description

The file_utils module is a widely-used utility module that provides the foundation for data I/O operations across the entire NeMo Curator codebase. It abstracts filesystem operations through the fsspec library, enabling transparent support for local files, S3, GCS, Azure Blob Storage, and other storage backends.

Filesystem Operations:

• get_fs: Creates an fsspec.AbstractFileSystem instance from a path by detecting the protocol (e.g., "s3://", "gcs://", local path) using split_protocol and get_filesystem_class. Accepts optional storage options for authentication.

• is_not_empty: Checks if a path exists, is a directory, and has contents. Accepts either a pre-built filesystem object or storage options.

• delete_dir: Recursively deletes a directory if it exists. Guards against double-specification of fs and storage_options.

• create_or_overwrite_dir: Creates a directory, deleting all existing contents first if the directory is not empty. Logs a warning before deletion.

File Discovery:

• get_all_file_paths_under: Discovers and returns a sorted list of file paths under a given path, with optional recursion into subdirectories and extension filtering. Delegates to the internal _gather_file_records function.

• get_all_file_paths_and_size_under: Same as above but also returns file sizes, sorted by size in ascending order. Useful for load balancing across workers.

• _gather_file_records: Internal function that performs the actual file discovery using fsspec's expand_path and find methods. Handles both directory and file inputs, normalizes paths for remote URLs, and filters by allowed extensions.

• filter_files_by_extension: Filters a list of file paths by one or more extensions, ensuring extensions are properly dot-prefixed.

• _split_files_as_per_blocksize: Partitions a list of files (with sizes) into groups where each group does not exceed a maximum byte size. Used for balanced data loading.

Output Mode Handling:

• check_output_mode: Implements write mode logic for output directories with four modes:
  • "overwrite": Deletes existing directory contents before proceeding
  • "append": Raises NotImplementedError if not implemented by the caller
  • "error": Raises FileExistsError if the directory already exists
  • "ignore": Proceeds without any action
  • All modes create the output directory if it does not exist.

DataFrame Utilities:

• pandas_select_columns: Safely projects a Pandas DataFrame onto requested columns, logging warnings for missing columns and returning None if none of the requested columns exist.

Path Utilities:

• infer_protocol_from_paths: Infers the storage protocol from a list of paths by inspecting storage options, returning the first non-local protocol found (e.g., "s3", "gcs", "abfs").

• infer_dataset_name_from_path: Extracts a dataset name from either a local or cloud storage path, using the parent directory name or file stem as appropriate.

• check_disallowed_kwargs: Validates that a dictionary does not contain disallowed keys, optionally raising an error or logging a warning.

Security:

• tar_safe_extract: Safely extracts a tar file while preventing path traversal attacks. Validates all member paths against the extraction directory, blocks absolute paths, device files, and symlinks that point outside the extraction directory. Uses the internal _is_safe_path helper.

Usage

This module is used across the entire NeMo Curator codebase wherever file I/O is needed. Reader stages use it for file discovery, writer stages use it for output mode handling, pipeline planners use it for file partitioning by size, and download stages use it for safe archive extraction.

Code Reference

Source Location

• Repository: NeMo-Curator
• File: nemo_curator/utils/file_utils.py
• Lines: 1-450

Signature

FILETYPE_TO_DEFAULT_EXTENSIONS = {
    "parquet": [".parquet"],
    "jsonl": [".jsonl", ".json"],
    "megatron": [".bin", ".idx"],
}

def get_fs(path: str, storage_options: dict[str, str] | None = None) -> fsspec.AbstractFileSystem: ...
def is_not_empty(path: str, fs=None, storage_options=None) -> bool: ...
def delete_dir(path: str, fs=None, storage_options=None) -> None: ...
def create_or_overwrite_dir(path: str, fs=None, storage_options=None) -> None: ...
def filter_files_by_extension(files_list: list[str], keep_extensions: str | list[str]) -> list[str]: ...
def get_all_file_paths_under(path: str, recurse_subdirectories=False, keep_extensions=None, storage_options=None, fs=None) -> list[str]: ...
def get_all_file_paths_and_size_under(path: str, recurse_subdirectories=False, keep_extensions=None, storage_options=None, fs=None) -> list[tuple[str, int]]: ...
def infer_protocol_from_paths(paths: Iterable[str]) -> str | None: ...
def pandas_select_columns(df: pd.DataFrame, columns: list[str] | None, file_path: str) -> pd.DataFrame | None: ...
def check_output_mode(mode: Literal["overwrite", "append", "error", "ignore"], fs, path, append_mode_implemented=False) -> None: ...
def infer_dataset_name_from_path(path: str) -> str: ...
def check_disallowed_kwargs(kwargs: dict, disallowed_keys: list[str], raise_error: bool = True) -> None: ...
def tar_safe_extract(tar: tarfile.TarFile, path: str) -> None: ...

Import

from nemo_curator.utils.file_utils import (
    get_fs,
    get_all_file_paths_under,
    get_all_file_paths_and_size_under,
    filter_files_by_extension,
    check_output_mode,
    create_or_overwrite_dir,
    delete_dir,
    is_not_empty,
    infer_protocol_from_paths,
    pandas_select_columns,
    infer_dataset_name_from_path,
    tar_safe_extract,
)

I/O Contract

Inputs

Name	Type	Required	Description
path	str	Yes	File or directory path (local or cloud URL) for most functions.
storage_options	None	No	Storage-specific options (e.g., AWS credentials) passed to fsspec.
fs	None	No	Pre-built filesystem object. Mutually exclusive with storage_options.
recurse_subdirectories	bool	No (default: False)	Whether to recursively search subdirectories for file discovery functions.
keep_extensions	list[str] | None	No	File extensions to filter by (e.g., ".parquet", ".jsonl").
mode	Literal["overwrite", "append", "error", "ignore"]	Yes (for check_output_mode)	Output write mode for directory handling.
files_list	list[str]	Yes (for filter_files_by_extension)	List of file paths to filter.
columns	None	No	Column names for DataFrame projection in pandas_select_columns.
tar	tarfile.TarFile	Yes (for tar_safe_extract)	An open TarFile object to safely extract.

Outputs

Name	Type	Description
fs	fsspec.AbstractFileSystem	Filesystem instance returned by get_fs.
file_paths	list[str]	Sorted list of file paths from get_all_file_paths_under.
file_paths_and_sizes	list[tuple[str, int]]	List of (path, size) tuples sorted by size from get_all_file_paths_and_size_under.
filtered_files	list[str]	Filtered file paths from filter_files_by_extension.
protocol	None	Detected storage protocol from infer_protocol_from_paths.
dataset_name	str	Inferred dataset name from infer_dataset_name_from_path.
is_empty	bool	Whether the directory is not empty, from is_not_empty.
df	None	Projected DataFrame from pandas_select_columns.

Usage Examples

File Discovery

from nemo_curator.utils.file_utils import get_all_file_paths_under

# Find all parquet files under a directory
files = get_all_file_paths_under(
    path="/data/processed/",
    recurse_subdirectories=True,
    keep_extensions=[".parquet"],
)
print(f"Found {len(files)} parquet files")

# Works with cloud storage too
s3_files = get_all_file_paths_under(
    path="s3://my-bucket/data/",
    recurse_subdirectories=True,
    keep_extensions=".jsonl",
    storage_options={"key": "...", "secret": "..."},
)

Output Mode Handling

from nemo_curator.utils.file_utils import get_fs, check_output_mode

output_path = "/data/output/results/"
fs = get_fs(output_path)

# Overwrite mode: deletes existing directory contents
check_output_mode(mode="overwrite", fs=fs, path=output_path)

# Error mode: raises FileExistsError if directory exists
# check_output_mode(mode="error", fs=fs, path=output_path)

# Ignore mode: proceeds without modification
# check_output_mode(mode="ignore", fs=fs, path=output_path)

File Size Partitioning

from nemo_curator.utils.file_utils import get_all_file_paths_and_size_under

# Get files sorted by size for load balancing
files_with_sizes = get_all_file_paths_and_size_under(
    path="/data/input/",
    recurse_subdirectories=True,
    keep_extensions=".parquet",
)
for filepath, size_bytes in files_with_sizes[:5]:
    print(f"{filepath}: {size_bytes / 1024:.1f} KB")

Safe Tar Extraction

import tarfile
from nemo_curator.utils.file_utils import tar_safe_extract

# Safely extract a tar file, preventing path traversal attacks
with tarfile.open("/data/downloads/archive.tar.gz", "r:gz") as tar:
    tar_safe_extract(tar, path="/data/extracted/")

DataFrame Column Selection

import pandas as pd
from nemo_curator.utils.file_utils import pandas_select_columns

df = pd.read_parquet("/data/sample.parquet")

# Select only specific columns, with warnings for missing ones
result = pandas_select_columns(
    df=df,
    columns=["text", "id", "language"],
    file_path="/data/sample.parquet",
)
if result is not None:
    print(f"Selected columns: {list(result.columns)}")

Related Pages

• Environment:NVIDIA_NeMo_Curator_Python_Linux_Base