cursus.processing.numerical.streaming_numerical_imputation_processor

Streaming Numerical Imputation Processor - IterableDataset Support

This processor extends NumericalVariableImputationProcessor to support fitting from PipelineIterableDataset by accumulating statistics incrementally.

Key Features: - Single-pass streaming accumulation - Memory-efficient (no full dataset loading) - Online mean/median/mode computation - Same process()/transform() API as base processor - Batch fitting for multiple fields (10x faster than individual fitting)

class StreamingNumericalImputationProcessor(column_name, imputation_value=None, strategy=None)[source]

Bases: NumericalVariableImputationProcessor

Streaming-aware numerical imputation processor.

Extends NumericalVariableImputationProcessor to support fitting from PipelineIterableDataset by accumulating statistics incrementally.

Uses online algorithms for efficient single-pass statistics computation: - Mean: Running sum and count - Median: Approximate using sample collection (memory-limited) - Mode: Approximate using sample collection (memory-limited)

Examples

>>> # Create streaming processor
>>> proc = StreamingNumericalImputationProcessor(
...     column_name='age',
...     strategy='mean'
... )
>>>
>>> # Fit from streaming dataset
>>> proc.fit_streaming(train_iterable_dataset)
>>>
>>> # Use in pipeline (same API as base processor)
>>> dataset.add_pipeline('age', proc)
fit_streaming(dataset, field_names=None, strategy=None, max_samples=None, median_sample_limit=100000, show_progress=False)[source]

Fit imputation values from streaming dataset.

Supports both single-field and multi-field (batch) fitting: - Single-field: Uses self.column_name, returns self for chaining - Multi-field: Processes all fields in ONE pass, returns Dict[field_name -> imputation_value]

Parameters:

  • dataset (torch.utils.data.IterableDataset) – PipelineIterableDataset to stream from

  • field_names (List[str] | None) – Optional list of fields for batch fitting. If None, uses self.column_name

  • strategy (str | None) – Optional strategy override (‘mean’, ‘median’, ‘mode’). If None, uses self.strategy

  • max_samples (int | None) – Optional limit on samples processed (for early stopping)

  • median_sample_limit (int) – Max samples to keep for median/mode computation

  • show_progress (bool) – Whether to show progress bar (for batch mode)

Returns:

self (for method chaining) - Multi-field mode: Dict[field_name -> imputation_value]

Return type:

  • Single-field mode

Raises:

ValueError – If strategy is unknown

fit(X, y=None)[source]

Fit imputation value from data.

Automatically detects input type and delegates to appropriate method: - IterableDataset: uses fit_streaming() - Series/DataFrame: uses parent class fit()

Parameters:

  • X (Series | DataFrame | torch.utils.data.IterableDataset) – Series, DataFrame, or IterableDataset

  • y (Series | None) – Ignored (for sklearn compatibility)

Returns:

self (for method chaining)

Return type:

StreamingNumericalImputationProcessor