Chunked Processing

Process large Parquet files that exceed available RAM by reading and transforming data in chunks.

Polars Only

Chunked processing is designed for Polars DataFrames and Parquet files. DuckDB handles large datasets natively through its out-of-core execution engine — no chunking needed.

Overview

When working with files larger than available memory, use process_chunked() instead of process(). This method:

• Reads Parquet files in configurable chunks using lazy evaluation
• Optionally keeps related rows together using a partition key
• Validates that operations are compatible with chunked processing
• Returns a ChunkedProtocol with per-chunk statistics

from transformplan import TransformPlan, Col

plan = (
    TransformPlan()
    .col_rename(column="PatientID", new_name="patient_id")
    .rows_filter(Col("age") >= 18)
    .rows_unique(columns=["patient_id", "visit_date"])
)

# Process a large file in chunks
result, protocol = plan.process_chunked(
    source="patients_10gb.parquet",
    partition_key="patient_id",  # Keep patient rows together
    chunk_size=100_000,
)

protocol.print()

Operation Compatibility

Not all operations can be used with chunked processing. Operations are classified into three categories:

Chunkable Operations

These operations process each row independently and work with any chunk:

Category	Operations
Column	col_drop, col_rename, col_cast, col_reorder, col_select, col_duplicate, col_fill_null, col_drop_null, col_drop_zero, col_add, col_add_uuid, col_hash, col_coalesce
Math	math_add, math_subtract, math_multiply, math_divide, math_clamp, math_abs, math_round, math_set_min, math_set_max, math_add_columns, math_subtract_columns, math_multiply_columns, math_divide_columns, math_percent_of
String	str_replace, str_slice, str_truncate, str_lower, str_upper, str_strip, str_pad, str_split, str_concat, str_extract
Datetime	dt_year, dt_month, dt_day, dt_week, dt_quarter, dt_year_month, dt_quarter_year, dt_calendar_week, dt_parse, dt_format, dt_diff_days, dt_age_years, dt_is_between, dt_truncate
Map	map_values, map_discretize, map_bool_to_int, map_null_to_value, map_value_to_null, map_case, map_from_column
Rows	rows_filter, rows_drop, rows_flag, rows_explode, rows_drop_nulls, rows_melt

Group-Dependent Operations

These operations need all rows for a group together. They work with chunked processing only when partition_key includes their grouping columns:

Operation	Group Parameter	Requirement
rows_unique	columns	partition_key must contain columns
rows_deduplicate	columns	partition_key must contain columns
math_cumsum	group_by	partition_key must contain group_by
math_rank	group_by	partition_key must contain group_by
math_diff_from_agg	group_by	partition_key must contain group_by

Example: To use rows_unique(columns=["patient_id"]), you must set partition_key="patient_id" (or a list containing it).

Global Operations (Blocked)

These operations require the full dataset and cannot be used with chunked processing:

• rows_sort - Requires global ordering
• rows_pivot - Needs all values to determine output columns
• rows_sample - Random sampling requires full dataset
• rows_head - Requires global ordering
• rows_tail - Requires global ordering

Attempting to use these operations will raise a ChunkingError.

Validation

Before processing, validate that your pipeline is compatible with chunked processing:

# Validate without processing
validation = plan.validate_chunked(
    schema={"patient_id": pl.Utf8, "age": pl.Int64, "visit_date": pl.Date},
    partition_key="patient_id"
)

print(validation)
# Pipeline is compatible with chunked processing.

# Or validate with a sample DataFrame
validation = plan.validate_chunked(data=sample_df, partition_key="patient_id")

if not validation.is_valid:
    for error in validation.errors:
        print(f"Error: {error}")

ChunkedProtocol Class

ChunkedProtocol

ChunkedProtocol()

Protocol for tracking chunked processing with per-chunk information.

Tracks the overall processing as well as individual chunk statistics.

Attributes:

Name	Type	Description
VERSION		Protocol version string.

Initialize an empty ChunkedProtocol.

Source code in transformplan/chunking.py

def __init__(self) -> None:
    """Initialize an empty ChunkedProtocol."""
    self._chunks: list[ChunkInfo] = []
    self._source_path: str | None = None
    self._partition_key: list[str] | None = None
    self._chunk_size: int | None = None
    self._created_at: str = datetime.now(timezone.utc).isoformat()
    self._metadata: dict[str, Any] = {}
    self._operations: list[dict[str, Any]] = []

chunks property

chunks: list[ChunkInfo]

List of chunk information.

Returns:

Type	Description
list[ChunkInfo]	List of ChunkInfo instances.

total_input_rows property

total_input_rows: int

Total rows across all input chunks.

Returns:

Type	Description
int	Sum of input rows.

total_output_rows property

total_output_rows: int

Total rows across all output chunks.

Returns:

Type	Description
int	Sum of output rows.

total_elapsed_seconds property

total_elapsed_seconds: float

Total processing time across all chunks.

Returns:

Type	Description
float	Sum of elapsed seconds.

num_chunks property

num_chunks: int

Number of chunks processed.

Returns:

Type	Description
int	Count of chunks.

metadata property

metadata: dict[str, Any]

Protocol metadata.

Returns:

Type	Description
dict[str, Any]	Dictionary of metadata.

set_source

set_source(path: str, partition_key: list[str] | None, chunk_size: int) -> None

Set source file information.

Source code in transformplan/chunking.py

def set_source(
    self,
    path: str,
    partition_key: list[str] | None,
    chunk_size: int,
) -> None:
    """Set source file information."""
    self._source_path = path
    self._partition_key = partition_key
    self._chunk_size = chunk_size

set_operations

set_operations(operations: list[dict[str, Any]]) -> None

Record the operations that were applied.

Source code in transformplan/chunking.py

def set_operations(self, operations: list[dict[str, Any]]) -> None:
    """Record the operations that were applied."""
    self._operations = operations

set_metadata

set_metadata(**kwargs: Any) -> None

Set arbitrary metadata on the protocol.

Source code in transformplan/chunking.py

def set_metadata(self, **kwargs: Any) -> None:  # noqa: ANN401
    """Set arbitrary metadata on the protocol."""
    self._metadata.update(kwargs)

add_chunk

add_chunk(chunk_info: ChunkInfo) -> None

Add information about a processed chunk.

Source code in transformplan/chunking.py

def add_chunk(self, chunk_info: ChunkInfo) -> None:
    """Add information about a processed chunk."""
    self._chunks.append(chunk_info)

output_hash

output_hash() -> str

Compute a combined hash of all output chunk hashes.

Returns:

Type	Description
str	A 16-character hex hash of all chunk output hashes combined.

Source code in transformplan/chunking.py

def output_hash(self) -> str:
    """Compute a combined hash of all output chunk hashes.

    Returns:
        A 16-character hex hash of all chunk output hashes combined.
    """
    if not self._chunks:
        return ""
    combined = "|".join(c.output_hash for c in self._chunks)
    return hashlib.sha256(combined.encode()).hexdigest()[:16]

to_dict

to_dict() -> dict[str, Any]

Serialize protocol to a dictionary.

Returns:

Type	Description
dict[str, Any]	Dictionary representation of the protocol.

Source code in transformplan/chunking.py

def to_dict(self) -> dict[str, Any]:
    """Serialize protocol to a dictionary.

    Returns:
        Dictionary representation of the protocol.
    """
    return {
        "version": self.VERSION,
        "created_at": self._created_at,
        "metadata": self._metadata,
        "source": {
            "path": self._source_path,
            "partition_key": self._partition_key,
            "chunk_size": self._chunk_size,
        },
        "operations": self._operations,
        "summary": {
            "num_chunks": self.num_chunks,
            "total_input_rows": self.total_input_rows,
            "total_output_rows": self.total_output_rows,
            "total_elapsed_seconds": round(self.total_elapsed_seconds, 4),
            "output_hash": self.output_hash(),
        },
        "chunks": [
            {
                "chunk_index": c.chunk_index,
                "input_rows": c.input_rows,
                "output_rows": c.output_rows,
                "input_hash": c.input_hash,
                "output_hash": c.output_hash,
                "elapsed_seconds": round(c.elapsed_seconds, 4),
            }
            for c in self._chunks
        ],
    }

from_dict classmethod

from_dict(data: dict[str, Any]) -> ChunkedProtocol

Deserialize protocol from a dictionary.

Returns:

Type	Description
ChunkedProtocol	ChunkedProtocol instance.

Source code in transformplan/chunking.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> ChunkedProtocol:
    """Deserialize protocol from a dictionary.

    Returns:
        ChunkedProtocol instance.
    """
    protocol = cls()
    protocol._created_at = data.get("created_at", protocol._created_at)
    protocol._metadata = data.get("metadata", {})

    source = data.get("source", {})
    protocol._source_path = source.get("path")
    protocol._partition_key = source.get("partition_key")
    protocol._chunk_size = source.get("chunk_size")

    protocol._operations = data.get("operations", [])

    for chunk_data in data.get("chunks", []):
        protocol._chunks.append(
            ChunkInfo(
                chunk_index=chunk_data["chunk_index"],
                input_rows=chunk_data["input_rows"],
                output_rows=chunk_data["output_rows"],
                input_hash=chunk_data["input_hash"],
                output_hash=chunk_data["output_hash"],
                elapsed_seconds=chunk_data["elapsed_seconds"],
            )
        )

    return protocol

to_json

to_json(path: str | Path | None = None, indent: int = 2) -> str

Serialize protocol to JSON.

Parameters:

Name	Type	Description	Default
path	str | Path | None	Optional file path to write to.	None
indent	int	JSON indentation level.	2

Returns:

Type	Description
str	JSON string.

Source code in transformplan/chunking.py

def to_json(self, path: str | Path | None = None, indent: int = 2) -> str:
    """Serialize protocol to JSON.

    Args:
        path: Optional file path to write to.
        indent: JSON indentation level.

    Returns:
        JSON string.
    """
    json_str = json.dumps(self.to_dict(), indent=indent)

    if path is not None:
        Path(path).write_text(json_str)

    return json_str

from_json classmethod

from_json(source: str | Path) -> ChunkedProtocol

Deserialize protocol from JSON.

Parameters:

Name	Type	Description	Default
source	str | Path	Either a JSON string or a path to a JSON file.	required

Returns:

Type	Description
ChunkedProtocol	ChunkedProtocol instance.

Source code in transformplan/chunking.py

@classmethod
def from_json(cls, source: str | Path) -> ChunkedProtocol:
    """Deserialize protocol from JSON.

    Args:
        source: Either a JSON string or a path to a JSON file.

    Returns:
        ChunkedProtocol instance.
    """
    if isinstance(source, Path) or not source.strip().startswith("{"):
        content = Path(source).read_text()
    else:
        content = source

    return cls.from_dict(json.loads(content))

summary

summary() -> str

Generate a human-readable summary of the chunked processing.

Returns:

Type	Description
str	Formatted string summary of the protocol.

Source code in transformplan/chunking.py

def summary(self) -> str:
    """Generate a human-readable summary of the chunked processing.

    Returns:
        Formatted string summary of the protocol.
    """
    lines = [
        "=" * 70,
        "CHUNKED PROCESSING PROTOCOL",
        "=" * 70,
    ]

    if self._metadata:
        for key, value in self._metadata.items():
            lines.append(f"{key}: {value}")
        lines.append("-" * 70)

    # Source info
    if self._source_path:
        lines.append(f"Source: {self._source_path}")
    if self._partition_key:
        lines.append(f"Partition key: {self._partition_key}")
    if self._chunk_size:
        lines.append(f"Target chunk size: {self._chunk_size:,}")
    lines.extend(
        [
            "-" * 70,
            f"Chunks processed: {self.num_chunks}",
            f"Total input rows: {self.total_input_rows:,}",
            f"Total output rows: {self.total_output_rows:,}",
        ]
    )
    rows_diff = self.total_output_rows - self.total_input_rows
    if rows_diff != 0:
        lines.append(f"Row change: {rows_diff:+,}")
    lines.append(f"Total time: {self.total_elapsed_seconds:.4f}s")
    if self.num_chunks > 0:
        avg_time = self.total_elapsed_seconds / self.num_chunks
        lines.append(f"Avg time per chunk: {avg_time:.4f}s")
    lines.extend((f"Output hash: {self.output_hash()}", "-" * 70))

    # Per-chunk details
    if self._chunks:
        lines.extend(
            (
                "",
                f"{'#':<6} {'Input':<12} {'Output':<12} {'Change':<10} {'Time':<10} {'Hash':<16}",
                "-" * 70,
            )
        )

        for chunk in self._chunks:
            idx = str(chunk.chunk_index)
            input_rows = f"{chunk.input_rows:,}"
            output_rows = f"{chunk.output_rows:,}"
            change = chunk.output_rows - chunk.input_rows
            change_str = f"{change:+,}" if change != 0 else "-"
            time_str = f"{chunk.elapsed_seconds:.4f}s"
            hash_str = chunk.output_hash

            lines.append(
                f"{idx:<6} {input_rows:<12} {output_rows:<12} {change_str:<10} {time_str:<10} {hash_str:<16}"
            )

    lines.append("=" * 70)
    return "\n".join(lines)

print

print() -> None

Print the protocol summary to stdout.

Source code in transformplan/chunking.py

def print(self) -> None:
    """Print the protocol summary to stdout."""
    print(self.summary())  # noqa: T201

ChunkValidationResult Class

ChunkValidationResult dataclass

ChunkValidationResult(
    is_valid: bool,
    errors: list[str] = list(),
    warnings: list[str] = list(),
    global_operations: list[str] = list(),
    group_dependent_ops: list[tuple[str, list[str] | None]] = list(),
)

Result of validating a pipeline for chunked processing.

Attributes:

Name	Type	Description
is_valid	bool	Whether the pipeline can be processed in chunks.
errors	list[str]	List of error messages explaining incompatibilities.
warnings	list[str]	List of warning messages (non-blocking).
global_operations	list[str]	Names of operations that require full dataset.
group_dependent_ops	list[tuple[str, list[str] | None]]	List of (operation, columns) for group-dependent ops.

is_valid instance-attribute

is_valid: bool

errors class-attribute instance-attribute

errors: list[str] = field(default_factory=list)

warnings class-attribute instance-attribute

warnings: list[str] = field(default_factory=list)

global_operations class-attribute instance-attribute

global_operations: list[str] = field(default_factory=list)

group_dependent_ops class-attribute instance-attribute

group_dependent_ops: list[tuple[str, list[str] | None]] = field(
    default_factory=list
)

ChunkingError Exception

ChunkingError

ChunkingError(
    message: str, validation_result: ChunkValidationResult | None = None
)

Bases: Exception

Raised when a pipeline is incompatible with chunked processing.

Attributes:

Name	Type	Description
validation_result		The validation result containing error details.

Initialize ChunkingError with message and optional validation result.

Source code in transformplan/chunking.py

def __init__(
    self, message: str, validation_result: ChunkValidationResult | None = None
) -> None:
    """Initialize ChunkingError with message and optional validation result."""
    super().__init__(message)
    self.validation_result = validation_result

Example: Processing Patient Records

import polars as pl
from transformplan import TransformPlan, Col, ChunkingError

# Build pipeline with group-dependent operation
plan = (
    TransformPlan()
    .col_rename(column="PatientID", new_name="patient_id")
    .dt_age_years(birth_column="date_of_birth", new_column="age")
    .rows_filter(Col("age") >= 18)
    .rows_unique(columns=["patient_id", "visit_date"])  # Needs partition key
    .col_drop("date_of_birth")
)

# Validate first
validation = plan.validate_chunked(
    schema={
        "PatientID": pl.Utf8,
        "date_of_birth": pl.Date,
        "visit_date": pl.Date,
        "diagnosis": pl.Utf8,
    },
    partition_key="PatientID"
)

if validation.is_valid:
    # Process the large file
    result, protocol = plan.process_chunked(
        source="patients_archive.parquet",
        partition_key="PatientID",
        chunk_size=50_000,
    )

    # View processing summary
    protocol.print()

    # Save audit trail
    protocol.to_json("chunked_audit.json")
else:
    print("Pipeline incompatible with chunking:")
    for error in validation.errors:
        print(f"  - {error}")

Example Output

======================================================================
CHUNKED PROCESSING PROTOCOL
======================================================================
Source: patients_archive.parquet
Partition key: ['PatientID']
Target chunk size: 50,000
----------------------------------------------------------------------
Chunks processed: 24
Total input rows: 1,187,432
Total output rows: 892,156
Row change: -295,276
Total time: 12.4523s
Avg time per chunk: 0.5188s
Output hash: 7a3b2c1d4e5f6789
----------------------------------------------------------------------

#      Input        Output       Change     Time       Hash
----------------------------------------------------------------------
0      49,832       37,291       -12,541    0.4821s    a1b2c3d4e5f67890
1      50,127       38,456       -11,671    0.5123s    b2c3d4e5f6789012
2      49,956       37,892       -12,064    0.4956s    c3d4e5f678901234
...
======================================================================