Struct StreamTransformer 

pub struct StreamTransformer { /* private fields */ }

Memory-efficient data transformer for streaming

Implementations

impl StreamTransformer

pub fn new() -> Self

Create a new stream transformer

Examples found in repository

examples/datasets_streaming_demo.rs (line 225)

fn demonstrate_stream_transformations() -> Result<(), Box<dyn std::error::Error>> {
    println!("🔄 STREAM TRANSFORMATIONS");
    println!("{}", "-".repeat(40));

    // Create a transformer pipeline
    let transformer = StreamTransformer::new()
        .add_standard_scaling()
        .add_missing_value_imputation();

    println!("Created transformation pipeline:");
    println!("  1. Standard scaling (z-score normalization)");
    println!("  2. Missing value imputation");

    let config = StreamConfig {
        chunk_size: 2000,
        buffer_size: 2,
        max_chunks: Some(5),
        ..Default::default()
    };

    let mut stream = stream_regression(10_000, 15, config)?;
    let mut transformed_chunks = 0;

    println!("\nProcessing and transforming chunks...");

    while let Some(mut chunk) = stream.next_chunk()? {
        println!("  Processing chunk {}", chunk.chunk_index + 1);

        // Show statistics before transformation
        let data_mean_before = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let data_std_before = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        println!(
            "    Before: mean = {:.3}, std = {:.3}",
            data_mean_before[0], data_std_before[0]
        );

        // Apply transformations
        transformer.transform_chunk(&mut chunk)?;

        // Show statistics after transformation
        let data_mean_after = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let data_std_after = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        println!(
            "    After:  mean = {:.3}, std = {:.3}",
            data_mean_after[0], data_std_after[0]
        );

        transformed_chunks += 1;

        if chunk.is_last {
            break;
        }
    }

    println!("\nTransformation Summary:");
    println!("  Chunks processed: {transformed_chunks}");
    println!("  Each chunk was transformed independently");
    println!("  Memory-efficient: only one chunk in memory at a time");

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_parallel_processing() -> Result<(), Box<dyn std::error::Error>> {
    println!("⚡ PARALLEL STREAM PROCESSING");
    println!("{}", "-".repeat(40));

    let config = StreamConfig {
        chunk_size: 1500,
        buffer_size: 4,
        num_workers: 4,
        max_chunks: Some(8),
        ..Default::default()
    };

    println!("Parallel processing configuration:");
    println!("  Workers: {}", config.num_workers);
    println!("  Chunk size: {}", config.chunk_size);
    println!("  Buffer size: {}", config.buffer_size);

    // Create a simple processor that computes statistics
    let _processor: StreamProcessor<DataChunk> = StreamProcessor::new(config.clone());

    // Define a processing function
    let compute_stats = |chunk: DataChunk| -> Result<
        HashMap<String, f64>,
        Box<dyn std::error::Error + Send + Sync>,
    > {
        let mut stats = HashMap::new();

        // Compute basic statistics
        let mean = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let std = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        stats.insert("mean_feature_0".to_string(), mean[0]);
        stats.insert("std_feature_0".to_string(), std[0]);
        stats.insert("n_samples".to_string(), chunk.n_samples() as f64);
        stats.insert("chunk_index".to_string(), chunk.chunk_index as f64);

        // Simulate some computation time
        std::thread::sleep(std::time::Duration::from_millis(100));

        Ok(stats)
    };

    println!("\nProcessing stream with parallel workers...");
    let start_time = Instant::now();

    let stream = stream_classification(12_000, 10, 3, config)?;

    // For demonstration, we'll process chunks sequentially with timing
    // In a real implementation, you'd use the processor.process_parallel method
    let mut stream_iter = stream;
    let mut chunk_results = Vec::new();

    while let Some(chunk) = stream_iter.next_chunk()? {
        let chunk_start = Instant::now();
        let chunk_id = chunk.chunk_index;
        let chunk_samples = chunk.n_samples();

        // Process chunk
        let stats = compute_stats(chunk)
            .map_err(|e| -> Box<dyn std::error::Error> { Box::new(std::io::Error::other(e)) })?;
        let chunk_time = chunk_start.elapsed();

        println!(
            "  Chunk {}: {} samples, {:.2}ms",
            chunk_id + 1,
            chunk_samples,
            chunk_time.as_millis()
        );

        chunk_results.push(stats);

        if chunk_results.len() >= 8 {
            break;
        }
    }

    let total_time = start_time.elapsed();

    println!("\nParallel Processing Results:");
    println!("  Total chunks: {}", chunk_results.len());
    println!("  Total time: {:.2}s", total_time.as_secs_f64());
    println!(
        "  Average time per chunk: {:.2}ms",
        total_time.as_millis() as f64 / chunk_results.len() as f64
    );

    // Aggregate statistics
    let total_samples: f64 = chunk_results
        .iter()
        .map(|stats| stats.get("n_samples").unwrap_or(&0.0))
        .sum();

    println!("  Total samples processed: {total_samples}");
    println!(
        "  Throughput: {:.1} samples/s",
        total_samples / total_time.as_secs_f64()
    );

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_performance_comparison() -> Result<(), Box<dyn std::error::Error>> {
    println!("📊 PERFORMANCE COMPARISON");
    println!("{}", "-".repeat(40));

    let dataset_sizes = vec![10_000, 50_000, 100_000];
    let chunk_sizes = vec![1_000, 5_000, 10_000];

    println!("Comparing streaming performance across different configurations:");
    println!();

    for &datasetsize in &dataset_sizes {
        println!("Dataset size: {datasetsize} samples");

        for &chunksize in &chunk_sizes {
            let config = StreamConfig {
                chunk_size: chunksize,
                buffer_size: 3,
                num_workers: 2,
                max_chunks: Some(datasetsize / chunksize),
                ..Default::default()
            };

            let start_time = Instant::now();
            let mut stream = stream_regression(datasetsize, 20, config)?;

            let mut processed_samples = 0;
            let mut processed_chunks = 0;

            while let Some(chunk) = stream.next_chunk()? {
                processed_samples += chunk.n_samples();
                processed_chunks += 1;

                // Simulate minimal processing
                let _stats = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0));

                if chunk.is_last || processed_samples >= datasetsize {
                    break;
                }
            }

            let duration = start_time.elapsed();
            let throughput = processed_samples as f64 / duration.as_secs_f64();

            println!(
                "  Chunk size {}: {:.2}s ({:.1} samples/s, {} chunks)",
                chunksize,
                duration.as_secs_f64(),
                throughput,
                processed_chunks
            );
        }
        println!();
    }

    println!("Performance Insights:");
    println!("  • Larger chunks = fewer iterations, better throughput");
    println!("  • Smaller chunks = lower memory usage, more responsive");
    println!("  • Optimal chunk size depends on memory constraints and processing complexity");

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_real_world_scenarios() -> Result<(), Box<dyn std::error::Error>> {
    println!("🌍 REAL-WORLD STREAMING SCENARIOS");
    println!("{}", "-".repeat(40));

    // Scenario 1: Training on large dataset with limited memory
    println!("Scenario 1: Large dataset training with memory constraints");
    simulate_training_scenario()?;

    // Scenario 2: Data preprocessing pipeline
    println!("\nScenario 2: Data preprocessing pipeline");
    simulate_preprocessing_pipeline()?;

    // Scenario 3: Model evaluation on large test set
    println!("\nScenario 3: Model evaluation on large test set");
    simulate_model_evaluation()?;

    println!();
    Ok(())
}

#[allow(dead_code)]
fn simulate_training_scenario() -> Result<(), Box<dyn std::error::Error>> {
    println!("  • Dataset: 500K samples, 100 features");
    println!("  • Memory limit: 200MB");
    println!("  • Goal: Train incrementally using mini-batches");

    let config = StreamConfig {
        chunk_size: 5_000, // Mini-batch size
        buffer_size: 2,    // Keep memory low
        memory_limit_mb: Some(200),
        max_chunks: Some(10), // Simulate partial processing
        ..Default::default()
    };

    let mut stream = stream_classification(500_000, 100, 10, config)?;
    let mut total_batches = 0;
    let mut total_samples = 0;

    let start_time = Instant::now();

    while let Some(chunk) = stream.next_chunk()? {
        // Simulate training on mini-batch
        let batchsize = chunk.n_samples();

        // Simulate gradient computation time
        std::thread::sleep(std::time::Duration::from_millis(20));

        total_batches += 1;
        total_samples += batchsize;

        if total_batches % 3 == 0 {
            println!("    Processed {total_batches} batches ({total_samples} samples)");
        }

        if chunk.is_last {
            break;
        }
    }

    let duration = start_time.elapsed();
    println!(
        "  ✅ Training simulation: {} batches, {:.2}s",
        total_batches,
        duration.as_secs_f64()
    );

    Ok(())
}

#[allow(dead_code)]
fn simulate_preprocessing_pipeline() -> Result<(), Box<dyn std::error::Error>> {
    println!("  • Raw data → Clean → Scale → Feature selection");
    println!("  • Process 200K samples in chunks");

    let config = StreamConfig {
        chunk_size: 8_000,
        buffer_size: 3,
        max_chunks: Some(5),
        ..Default::default()
    };

    let transformer = StreamTransformer::new()
        .add_missing_value_imputation()
        .add_standard_scaling();

    let mut stream = stream_regression(200_000, 50, config)?;
    let mut processed_chunks = 0;

    while let Some(mut chunk) = stream.next_chunk()? {
        // Step 1: Clean data (remove outliers, handle missing values)
        transformer.transform_chunk(&mut chunk)?;

        // Step 2: Feature selection (simulate by keeping first 30 features)
        let selecteddata = chunk
            .data
            .slice(scirs2_core::ndarray::s![.., ..30])
            .to_owned();

        processed_chunks += 1;
        println!(
            "    Chunk {}: {} → {} features",
            processed_chunks,
            chunk.n_features(),
            selecteddata.ncols()
        );

        if chunk.is_last {
            break;
        }
    }

    println!("  ✅ Preprocessing pipeline: {processed_chunks} chunks processed");

    Ok(())
}

pub fn add_transform<F>(self, transform: F) -> Self

where F: Fn(&mut DataChunk) -> Result<()> + Send + Sync + 'static,

Add a transformation function

pub fn transform_chunk(&self, chunk: &mut DataChunk) -> Result<()>

Apply all transformations to a chunk

Examples found in repository

examples/datasets_streaming_demo.rs (line 258)

fn demonstrate_stream_transformations() -> Result<(), Box<dyn std::error::Error>> {
    println!("🔄 STREAM TRANSFORMATIONS");
    println!("{}", "-".repeat(40));

    // Create a transformer pipeline
    let transformer = StreamTransformer::new()
        .add_standard_scaling()
        .add_missing_value_imputation();

    println!("Created transformation pipeline:");
    println!("  1. Standard scaling (z-score normalization)");
    println!("  2. Missing value imputation");

    let config = StreamConfig {
        chunk_size: 2000,
        buffer_size: 2,
        max_chunks: Some(5),
        ..Default::default()
    };

    let mut stream = stream_regression(10_000, 15, config)?;
    let mut transformed_chunks = 0;

    println!("\nProcessing and transforming chunks...");

    while let Some(mut chunk) = stream.next_chunk()? {
        println!("  Processing chunk {}", chunk.chunk_index + 1);

        // Show statistics before transformation
        let data_mean_before = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let data_std_before = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        println!(
            "    Before: mean = {:.3}, std = {:.3}",
            data_mean_before[0], data_std_before[0]
        );

        // Apply transformations
        transformer.transform_chunk(&mut chunk)?;

        // Show statistics after transformation
        let data_mean_after = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let data_std_after = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        println!(
            "    After:  mean = {:.3}, std = {:.3}",
            data_mean_after[0], data_std_after[0]
        );

        transformed_chunks += 1;

        if chunk.is_last {
            break;
        }
    }

    println!("\nTransformation Summary:");
    println!("  Chunks processed: {transformed_chunks}");
    println!("  Each chunk was transformed independently");
    println!("  Memory-efficient: only one chunk in memory at a time");

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_parallel_processing() -> Result<(), Box<dyn std::error::Error>> {
    println!("⚡ PARALLEL STREAM PROCESSING");
    println!("{}", "-".repeat(40));

    let config = StreamConfig {
        chunk_size: 1500,
        buffer_size: 4,
        num_workers: 4,
        max_chunks: Some(8),
        ..Default::default()
    };

    println!("Parallel processing configuration:");
    println!("  Workers: {}", config.num_workers);
    println!("  Chunk size: {}", config.chunk_size);
    println!("  Buffer size: {}", config.buffer_size);

    // Create a simple processor that computes statistics
    let _processor: StreamProcessor<DataChunk> = StreamProcessor::new(config.clone());

    // Define a processing function
    let compute_stats = |chunk: DataChunk| -> Result<
        HashMap<String, f64>,
        Box<dyn std::error::Error + Send + Sync>,
    > {
        let mut stats = HashMap::new();

        // Compute basic statistics
        let mean = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let std = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        stats.insert("mean_feature_0".to_string(), mean[0]);
        stats.insert("std_feature_0".to_string(), std[0]);
        stats.insert("n_samples".to_string(), chunk.n_samples() as f64);
        stats.insert("chunk_index".to_string(), chunk.chunk_index as f64);

        // Simulate some computation time
        std::thread::sleep(std::time::Duration::from_millis(100));

        Ok(stats)
    };

    println!("\nProcessing stream with parallel workers...");
    let start_time = Instant::now();

    let stream = stream_classification(12_000, 10, 3, config)?;

    // For demonstration, we'll process chunks sequentially with timing
    // In a real implementation, you'd use the processor.process_parallel method
    let mut stream_iter = stream;
    let mut chunk_results = Vec::new();

    while let Some(chunk) = stream_iter.next_chunk()? {
        let chunk_start = Instant::now();
        let chunk_id = chunk.chunk_index;
        let chunk_samples = chunk.n_samples();

        // Process chunk
        let stats = compute_stats(chunk)
            .map_err(|e| -> Box<dyn std::error::Error> { Box::new(std::io::Error::other(e)) })?;
        let chunk_time = chunk_start.elapsed();

        println!(
            "  Chunk {}: {} samples, {:.2}ms",
            chunk_id + 1,
            chunk_samples,
            chunk_time.as_millis()
        );

        chunk_results.push(stats);

        if chunk_results.len() >= 8 {
            break;
        }
    }

    let total_time = start_time.elapsed();

    println!("\nParallel Processing Results:");
    println!("  Total chunks: {}", chunk_results.len());
    println!("  Total time: {:.2}s", total_time.as_secs_f64());
    println!(
        "  Average time per chunk: {:.2}ms",
        total_time.as_millis() as f64 / chunk_results.len() as f64
    );

    // Aggregate statistics
    let total_samples: f64 = chunk_results
        .iter()
        .map(|stats| stats.get("n_samples").unwrap_or(&0.0))
        .sum();

    println!("  Total samples processed: {total_samples}");
    println!(
        "  Throughput: {:.1} samples/s",
        total_samples / total_time.as_secs_f64()
    );

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_performance_comparison() -> Result<(), Box<dyn std::error::Error>> {
    println!("📊 PERFORMANCE COMPARISON");
    println!("{}", "-".repeat(40));

    let dataset_sizes = vec![10_000, 50_000, 100_000];
    let chunk_sizes = vec![1_000, 5_000, 10_000];

    println!("Comparing streaming performance across different configurations:");
    println!();

    for &datasetsize in &dataset_sizes {
        println!("Dataset size: {datasetsize} samples");

        for &chunksize in &chunk_sizes {
            let config = StreamConfig {
                chunk_size: chunksize,
                buffer_size: 3,
                num_workers: 2,
                max_chunks: Some(datasetsize / chunksize),
                ..Default::default()
            };

            let start_time = Instant::now();
            let mut stream = stream_regression(datasetsize, 20, config)?;

            let mut processed_samples = 0;
            let mut processed_chunks = 0;

            while let Some(chunk) = stream.next_chunk()? {
                processed_samples += chunk.n_samples();
                processed_chunks += 1;

                // Simulate minimal processing
                let _stats = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0));

                if chunk.is_last || processed_samples >= datasetsize {
                    break;
                }
            }

            let duration = start_time.elapsed();
            let throughput = processed_samples as f64 / duration.as_secs_f64();

            println!(
                "  Chunk size {}: {:.2}s ({:.1} samples/s, {} chunks)",
                chunksize,
                duration.as_secs_f64(),
                throughput,
                processed_chunks
            );
        }
        println!();
    }

    println!("Performance Insights:");
    println!("  • Larger chunks = fewer iterations, better throughput");
    println!("  • Smaller chunks = lower memory usage, more responsive");
    println!("  • Optimal chunk size depends on memory constraints and processing complexity");

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_real_world_scenarios() -> Result<(), Box<dyn std::error::Error>> {
    println!("🌍 REAL-WORLD STREAMING SCENARIOS");
    println!("{}", "-".repeat(40));

    // Scenario 1: Training on large dataset with limited memory
    println!("Scenario 1: Large dataset training with memory constraints");
    simulate_training_scenario()?;

    // Scenario 2: Data preprocessing pipeline
    println!("\nScenario 2: Data preprocessing pipeline");
    simulate_preprocessing_pipeline()?;

    // Scenario 3: Model evaluation on large test set
    println!("\nScenario 3: Model evaluation on large test set");
    simulate_model_evaluation()?;

    println!();
    Ok(())
}

#[allow(dead_code)]
fn simulate_training_scenario() -> Result<(), Box<dyn std::error::Error>> {
    println!("  • Dataset: 500K samples, 100 features");
    println!("  • Memory limit: 200MB");
    println!("  • Goal: Train incrementally using mini-batches");

    let config = StreamConfig {
        chunk_size: 5_000, // Mini-batch size
        buffer_size: 2,    // Keep memory low
        memory_limit_mb: Some(200),
        max_chunks: Some(10), // Simulate partial processing
        ..Default::default()
    };

    let mut stream = stream_classification(500_000, 100, 10, config)?;
    let mut total_batches = 0;
    let mut total_samples = 0;

    let start_time = Instant::now();

    while let Some(chunk) = stream.next_chunk()? {
        // Simulate training on mini-batch
        let batchsize = chunk.n_samples();

        // Simulate gradient computation time
        std::thread::sleep(std::time::Duration::from_millis(20));

        total_batches += 1;
        total_samples += batchsize;

        if total_batches % 3 == 0 {
            println!("    Processed {total_batches} batches ({total_samples} samples)");
        }

        if chunk.is_last {
            break;
        }
    }

    let duration = start_time.elapsed();
    println!(
        "  ✅ Training simulation: {} batches, {:.2}s",
        total_batches,
        duration.as_secs_f64()
    );

    Ok(())
}

#[allow(dead_code)]
fn simulate_preprocessing_pipeline() -> Result<(), Box<dyn std::error::Error>> {
    println!("  • Raw data → Clean → Scale → Feature selection");
    println!("  • Process 200K samples in chunks");

    let config = StreamConfig {
        chunk_size: 8_000,
        buffer_size: 3,
        max_chunks: Some(5),
        ..Default::default()
    };

    let transformer = StreamTransformer::new()
        .add_missing_value_imputation()
        .add_standard_scaling();

    let mut stream = stream_regression(200_000, 50, config)?;
    let mut processed_chunks = 0;

    while let Some(mut chunk) = stream.next_chunk()? {
        // Step 1: Clean data (remove outliers, handle missing values)
        transformer.transform_chunk(&mut chunk)?;

        // Step 2: Feature selection (simulate by keeping first 30 features)
        let selecteddata = chunk
            .data
            .slice(scirs2_core::ndarray::s![.., ..30])
            .to_owned();

        processed_chunks += 1;
        println!(
            "    Chunk {}: {} → {} features",
            processed_chunks,
            chunk.n_features(),
            selecteddata.ncols()
        );

        if chunk.is_last {
            break;
        }
    }

    println!("  ✅ Preprocessing pipeline: {processed_chunks} chunks processed");

    Ok(())
}

pub fn add_standard_scaling(self) -> Self

Add standard scaling transformation

Examples found in repository

examples/datasets_streaming_demo.rs (line 226)

fn demonstrate_stream_transformations() -> Result<(), Box<dyn std::error::Error>> {
    println!("🔄 STREAM TRANSFORMATIONS");
    println!("{}", "-".repeat(40));

    // Create a transformer pipeline
    let transformer = StreamTransformer::new()
        .add_standard_scaling()
        .add_missing_value_imputation();

    println!("Created transformation pipeline:");
    println!("  1. Standard scaling (z-score normalization)");
    println!("  2. Missing value imputation");

    let config = StreamConfig {
        chunk_size: 2000,
        buffer_size: 2,
        max_chunks: Some(5),
        ..Default::default()
    };

    let mut stream = stream_regression(10_000, 15, config)?;
    let mut transformed_chunks = 0;

    println!("\nProcessing and transforming chunks...");

    while let Some(mut chunk) = stream.next_chunk()? {
        println!("  Processing chunk {}", chunk.chunk_index + 1);

        // Show statistics before transformation
        let data_mean_before = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let data_std_before = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        println!(
            "    Before: mean = {:.3}, std = {:.3}",
            data_mean_before[0], data_std_before[0]
        );

        // Apply transformations
        transformer.transform_chunk(&mut chunk)?;

        // Show statistics after transformation
        let data_mean_after = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let data_std_after = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        println!(
            "    After:  mean = {:.3}, std = {:.3}",
            data_mean_after[0], data_std_after[0]
        );

        transformed_chunks += 1;

        if chunk.is_last {
            break;
        }
    }

    println!("\nTransformation Summary:");
    println!("  Chunks processed: {transformed_chunks}");
    println!("  Each chunk was transformed independently");
    println!("  Memory-efficient: only one chunk in memory at a time");

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_parallel_processing() -> Result<(), Box<dyn std::error::Error>> {
    println!("⚡ PARALLEL STREAM PROCESSING");
    println!("{}", "-".repeat(40));

    let config = StreamConfig {
        chunk_size: 1500,
        buffer_size: 4,
        num_workers: 4,
        max_chunks: Some(8),
        ..Default::default()
    };

    println!("Parallel processing configuration:");
    println!("  Workers: {}", config.num_workers);
    println!("  Chunk size: {}", config.chunk_size);
    println!("  Buffer size: {}", config.buffer_size);

    // Create a simple processor that computes statistics
    let _processor: StreamProcessor<DataChunk> = StreamProcessor::new(config.clone());

    // Define a processing function
    let compute_stats = |chunk: DataChunk| -> Result<
        HashMap<String, f64>,
        Box<dyn std::error::Error + Send + Sync>,
    > {
        let mut stats = HashMap::new();

        // Compute basic statistics
        let mean = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let std = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        stats.insert("mean_feature_0".to_string(), mean[0]);
        stats.insert("std_feature_0".to_string(), std[0]);
        stats.insert("n_samples".to_string(), chunk.n_samples() as f64);
        stats.insert("chunk_index".to_string(), chunk.chunk_index as f64);

        // Simulate some computation time
        std::thread::sleep(std::time::Duration::from_millis(100));

        Ok(stats)
    };

    println!("\nProcessing stream with parallel workers...");
    let start_time = Instant::now();

    let stream = stream_classification(12_000, 10, 3, config)?;

    // For demonstration, we'll process chunks sequentially with timing
    // In a real implementation, you'd use the processor.process_parallel method
    let mut stream_iter = stream;
    let mut chunk_results = Vec::new();

    while let Some(chunk) = stream_iter.next_chunk()? {
        let chunk_start = Instant::now();
        let chunk_id = chunk.chunk_index;
        let chunk_samples = chunk.n_samples();

        // Process chunk
        let stats = compute_stats(chunk)
            .map_err(|e| -> Box<dyn std::error::Error> { Box::new(std::io::Error::other(e)) })?;
        let chunk_time = chunk_start.elapsed();

        println!(
            "  Chunk {}: {} samples, {:.2}ms",
            chunk_id + 1,
            chunk_samples,
            chunk_time.as_millis()
        );

        chunk_results.push(stats);

        if chunk_results.len() >= 8 {
            break;
        }
    }

    let total_time = start_time.elapsed();

    println!("\nParallel Processing Results:");
    println!("  Total chunks: {}", chunk_results.len());
    println!("  Total time: {:.2}s", total_time.as_secs_f64());
    println!(
        "  Average time per chunk: {:.2}ms",
        total_time.as_millis() as f64 / chunk_results.len() as f64
    );

    // Aggregate statistics
    let total_samples: f64 = chunk_results
        .iter()
        .map(|stats| stats.get("n_samples").unwrap_or(&0.0))
        .sum();

    println!("  Total samples processed: {total_samples}");
    println!(
        "  Throughput: {:.1} samples/s",
        total_samples / total_time.as_secs_f64()
    );

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_performance_comparison() -> Result<(), Box<dyn std::error::Error>> {
    println!("📊 PERFORMANCE COMPARISON");
    println!("{}", "-".repeat(40));

    let dataset_sizes = vec![10_000, 50_000, 100_000];
    let chunk_sizes = vec![1_000, 5_000, 10_000];

    println!("Comparing streaming performance across different configurations:");
    println!();

    for &datasetsize in &dataset_sizes {
        println!("Dataset size: {datasetsize} samples");

        for &chunksize in &chunk_sizes {
            let config = StreamConfig {
                chunk_size: chunksize,
                buffer_size: 3,
                num_workers: 2,
                max_chunks: Some(datasetsize / chunksize),
                ..Default::default()
            };

            let start_time = Instant::now();
            let mut stream = stream_regression(datasetsize, 20, config)?;

            let mut processed_samples = 0;
            let mut processed_chunks = 0;

            while let Some(chunk) = stream.next_chunk()? {
                processed_samples += chunk.n_samples();
                processed_chunks += 1;

                // Simulate minimal processing
                let _stats = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0));

                if chunk.is_last || processed_samples >= datasetsize {
                    break;
                }
            }

            let duration = start_time.elapsed();
            let throughput = processed_samples as f64 / duration.as_secs_f64();

            println!(
                "  Chunk size {}: {:.2}s ({:.1} samples/s, {} chunks)",
                chunksize,
                duration.as_secs_f64(),
                throughput,
                processed_chunks
            );
        }
        println!();
    }

    println!("Performance Insights:");
    println!("  • Larger chunks = fewer iterations, better throughput");
    println!("  • Smaller chunks = lower memory usage, more responsive");
    println!("  • Optimal chunk size depends on memory constraints and processing complexity");

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_real_world_scenarios() -> Result<(), Box<dyn std::error::Error>> {
    println!("🌍 REAL-WORLD STREAMING SCENARIOS");
    println!("{}", "-".repeat(40));

    // Scenario 1: Training on large dataset with limited memory
    println!("Scenario 1: Large dataset training with memory constraints");
    simulate_training_scenario()?;

    // Scenario 2: Data preprocessing pipeline
    println!("\nScenario 2: Data preprocessing pipeline");
    simulate_preprocessing_pipeline()?;

    // Scenario 3: Model evaluation on large test set
    println!("\nScenario 3: Model evaluation on large test set");
    simulate_model_evaluation()?;

    println!();
    Ok(())
}

#[allow(dead_code)]
fn simulate_training_scenario() -> Result<(), Box<dyn std::error::Error>> {
    println!("  • Dataset: 500K samples, 100 features");
    println!("  • Memory limit: 200MB");
    println!("  • Goal: Train incrementally using mini-batches");

    let config = StreamConfig {
        chunk_size: 5_000, // Mini-batch size
        buffer_size: 2,    // Keep memory low
        memory_limit_mb: Some(200),
        max_chunks: Some(10), // Simulate partial processing
        ..Default::default()
    };

    let mut stream = stream_classification(500_000, 100, 10, config)?;
    let mut total_batches = 0;
    let mut total_samples = 0;

    let start_time = Instant::now();

    while let Some(chunk) = stream.next_chunk()? {
        // Simulate training on mini-batch
        let batchsize = chunk.n_samples();

        // Simulate gradient computation time
        std::thread::sleep(std::time::Duration::from_millis(20));

        total_batches += 1;
        total_samples += batchsize;

        if total_batches % 3 == 0 {
            println!("    Processed {total_batches} batches ({total_samples} samples)");
        }

        if chunk.is_last {
            break;
        }
    }

    let duration = start_time.elapsed();
    println!(
        "  ✅ Training simulation: {} batches, {:.2}s",
        total_batches,
        duration.as_secs_f64()
    );

    Ok(())
}

#[allow(dead_code)]
fn simulate_preprocessing_pipeline() -> Result<(), Box<dyn std::error::Error>> {
    println!("  • Raw data → Clean → Scale → Feature selection");
    println!("  • Process 200K samples in chunks");

    let config = StreamConfig {
        chunk_size: 8_000,
        buffer_size: 3,
        max_chunks: Some(5),
        ..Default::default()
    };

    let transformer = StreamTransformer::new()
        .add_missing_value_imputation()
        .add_standard_scaling();

    let mut stream = stream_regression(200_000, 50, config)?;
    let mut processed_chunks = 0;

    while let Some(mut chunk) = stream.next_chunk()? {
        // Step 1: Clean data (remove outliers, handle missing values)
        transformer.transform_chunk(&mut chunk)?;

        // Step 2: Feature selection (simulate by keeping first 30 features)
        let selecteddata = chunk
            .data
            .slice(scirs2_core::ndarray::s![.., ..30])
            .to_owned();

        processed_chunks += 1;
        println!(
            "    Chunk {}: {} → {} features",
            processed_chunks,
            chunk.n_features(),
            selecteddata.ncols()
        );

        if chunk.is_last {
            break;
        }
    }

    println!("  ✅ Preprocessing pipeline: {processed_chunks} chunks processed");

    Ok(())
}

pub fn add_missing_value_imputation(self) -> Self

Add missing value imputation

Examples found in repository

examples/datasets_streaming_demo.rs (line 227)

fn demonstrate_stream_transformations() -> Result<(), Box<dyn std::error::Error>> {
    println!("🔄 STREAM TRANSFORMATIONS");
    println!("{}", "-".repeat(40));

    // Create a transformer pipeline
    let transformer = StreamTransformer::new()
        .add_standard_scaling()
        .add_missing_value_imputation();

    println!("Created transformation pipeline:");
    println!("  1. Standard scaling (z-score normalization)");
    println!("  2. Missing value imputation");

    let config = StreamConfig {
        chunk_size: 2000,
        buffer_size: 2,
        max_chunks: Some(5),
        ..Default::default()
    };

    let mut stream = stream_regression(10_000, 15, config)?;
    let mut transformed_chunks = 0;

    println!("\nProcessing and transforming chunks...");

    while let Some(mut chunk) = stream.next_chunk()? {
        println!("  Processing chunk {}", chunk.chunk_index + 1);

        // Show statistics before transformation
        let data_mean_before = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let data_std_before = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        println!(
            "    Before: mean = {:.3}, std = {:.3}",
            data_mean_before[0], data_std_before[0]
        );

        // Apply transformations
        transformer.transform_chunk(&mut chunk)?;

        // Show statistics after transformation
        let data_mean_after = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let data_std_after = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        println!(
            "    After:  mean = {:.3}, std = {:.3}",
            data_mean_after[0], data_std_after[0]
        );

        transformed_chunks += 1;

        if chunk.is_last {
            break;
        }
    }

    println!("\nTransformation Summary:");
    println!("  Chunks processed: {transformed_chunks}");
    println!("  Each chunk was transformed independently");
    println!("  Memory-efficient: only one chunk in memory at a time");

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_parallel_processing() -> Result<(), Box<dyn std::error::Error>> {
    println!("⚡ PARALLEL STREAM PROCESSING");
    println!("{}", "-".repeat(40));

    let config = StreamConfig {
        chunk_size: 1500,
        buffer_size: 4,
        num_workers: 4,
        max_chunks: Some(8),
        ..Default::default()
    };

    println!("Parallel processing configuration:");
    println!("  Workers: {}", config.num_workers);
    println!("  Chunk size: {}", config.chunk_size);
    println!("  Buffer size: {}", config.buffer_size);

    // Create a simple processor that computes statistics
    let _processor: StreamProcessor<DataChunk> = StreamProcessor::new(config.clone());

    // Define a processing function
    let compute_stats = |chunk: DataChunk| -> Result<
        HashMap<String, f64>,
        Box<dyn std::error::Error + Send + Sync>,
    > {
        let mut stats = HashMap::new();

        // Compute basic statistics
        let mean = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let std = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        stats.insert("mean_feature_0".to_string(), mean[0]);
        stats.insert("std_feature_0".to_string(), std[0]);
        stats.insert("n_samples".to_string(), chunk.n_samples() as f64);
        stats.insert("chunk_index".to_string(), chunk.chunk_index as f64);

        // Simulate some computation time
        std::thread::sleep(std::time::Duration::from_millis(100));

        Ok(stats)
    };

    println!("\nProcessing stream with parallel workers...");
    let start_time = Instant::now();

    let stream = stream_classification(12_000, 10, 3, config)?;

    // For demonstration, we'll process chunks sequentially with timing
    // In a real implementation, you'd use the processor.process_parallel method
    let mut stream_iter = stream;
    let mut chunk_results = Vec::new();

    while let Some(chunk) = stream_iter.next_chunk()? {
        let chunk_start = Instant::now();
        let chunk_id = chunk.chunk_index;
        let chunk_samples = chunk.n_samples();

        // Process chunk
        let stats = compute_stats(chunk)
            .map_err(|e| -> Box<dyn std::error::Error> { Box::new(std::io::Error::other(e)) })?;
        let chunk_time = chunk_start.elapsed();

        println!(
            "  Chunk {}: {} samples, {:.2}ms",
            chunk_id + 1,
            chunk_samples,
            chunk_time.as_millis()
        );

        chunk_results.push(stats);

        if chunk_results.len() >= 8 {
            break;
        }
    }

    let total_time = start_time.elapsed();

    println!("\nParallel Processing Results:");
    println!("  Total chunks: {}", chunk_results.len());
    println!("  Total time: {:.2}s", total_time.as_secs_f64());
    println!(
        "  Average time per chunk: {:.2}ms",
        total_time.as_millis() as f64 / chunk_results.len() as f64
    );

    // Aggregate statistics
    let total_samples: f64 = chunk_results
        .iter()
        .map(|stats| stats.get("n_samples").unwrap_or(&0.0))
        .sum();

    println!("  Total samples processed: {total_samples}");
    println!(
        "  Throughput: {:.1} samples/s",
        total_samples / total_time.as_secs_f64()
    );

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_performance_comparison() -> Result<(), Box<dyn std::error::Error>> {
    println!("📊 PERFORMANCE COMPARISON");
    println!("{}", "-".repeat(40));

    let dataset_sizes = vec![10_000, 50_000, 100_000];
    let chunk_sizes = vec![1_000, 5_000, 10_000];

    println!("Comparing streaming performance across different configurations:");
    println!();

    for &datasetsize in &dataset_sizes {
        println!("Dataset size: {datasetsize} samples");

        for &chunksize in &chunk_sizes {
            let config = StreamConfig {
                chunk_size: chunksize,
                buffer_size: 3,
                num_workers: 2,
                max_chunks: Some(datasetsize / chunksize),
                ..Default::default()
            };

            let start_time = Instant::now();
            let mut stream = stream_regression(datasetsize, 20, config)?;

            let mut processed_samples = 0;
            let mut processed_chunks = 0;

            while let Some(chunk) = stream.next_chunk()? {
                processed_samples += chunk.n_samples();
                processed_chunks += 1;

                // Simulate minimal processing
                let _stats = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0));

                if chunk.is_last || processed_samples >= datasetsize {
                    break;
                }
            }

            let duration = start_time.elapsed();
            let throughput = processed_samples as f64 / duration.as_secs_f64();

            println!(
                "  Chunk size {}: {:.2}s ({:.1} samples/s, {} chunks)",
                chunksize,
                duration.as_secs_f64(),
                throughput,
                processed_chunks
            );
        }
        println!();
    }

    println!("Performance Insights:");
    println!("  • Larger chunks = fewer iterations, better throughput");
    println!("  • Smaller chunks = lower memory usage, more responsive");
    println!("  • Optimal chunk size depends on memory constraints and processing complexity");

    println!();
    Ok(())
}

#[allow(dead_code)]
fn demonstrate_real_world_scenarios() -> Result<(), Box<dyn std::error::Error>> {
    println!("🌍 REAL-WORLD STREAMING SCENARIOS");
    println!("{}", "-".repeat(40));

    // Scenario 1: Training on large dataset with limited memory
    println!("Scenario 1: Large dataset training with memory constraints");
    simulate_training_scenario()?;

    // Scenario 2: Data preprocessing pipeline
    println!("\nScenario 2: Data preprocessing pipeline");
    simulate_preprocessing_pipeline()?;

    // Scenario 3: Model evaluation on large test set
    println!("\nScenario 3: Model evaluation on large test set");
    simulate_model_evaluation()?;

    println!();
    Ok(())
}

#[allow(dead_code)]
fn simulate_training_scenario() -> Result<(), Box<dyn std::error::Error>> {
    println!("  • Dataset: 500K samples, 100 features");
    println!("  • Memory limit: 200MB");
    println!("  • Goal: Train incrementally using mini-batches");

    let config = StreamConfig {
        chunk_size: 5_000, // Mini-batch size
        buffer_size: 2,    // Keep memory low
        memory_limit_mb: Some(200),
        max_chunks: Some(10), // Simulate partial processing
        ..Default::default()
    };

    let mut stream = stream_classification(500_000, 100, 10, config)?;
    let mut total_batches = 0;
    let mut total_samples = 0;

    let start_time = Instant::now();

    while let Some(chunk) = stream.next_chunk()? {
        // Simulate training on mini-batch
        let batchsize = chunk.n_samples();

        // Simulate gradient computation time
        std::thread::sleep(std::time::Duration::from_millis(20));

        total_batches += 1;
        total_samples += batchsize;

        if total_batches % 3 == 0 {
            println!("    Processed {total_batches} batches ({total_samples} samples)");
        }

        if chunk.is_last {
            break;
        }
    }

    let duration = start_time.elapsed();
    println!(
        "  ✅ Training simulation: {} batches, {:.2}s",
        total_batches,
        duration.as_secs_f64()
    );

    Ok(())
}

#[allow(dead_code)]
fn simulate_preprocessing_pipeline() -> Result<(), Box<dyn std::error::Error>> {
    println!("  • Raw data → Clean → Scale → Feature selection");
    println!("  • Process 200K samples in chunks");

    let config = StreamConfig {
        chunk_size: 8_000,
        buffer_size: 3,
        max_chunks: Some(5),
        ..Default::default()
    };

    let transformer = StreamTransformer::new()
        .add_missing_value_imputation()
        .add_standard_scaling();

    let mut stream = stream_regression(200_000, 50, config)?;
    let mut processed_chunks = 0;

    while let Some(mut chunk) = stream.next_chunk()? {
        // Step 1: Clean data (remove outliers, handle missing values)
        transformer.transform_chunk(&mut chunk)?;

        // Step 2: Feature selection (simulate by keeping first 30 features)
        let selecteddata = chunk
            .data
            .slice(scirs2_core::ndarray::s![.., ..30])
            .to_owned();

        processed_chunks += 1;
        println!(
            "    Chunk {}: {} → {} features",
            processed_chunks,
            chunk.n_features(),
            selecteddata.ncols()
        );

        if chunk.is_last {
            break;
        }
    }

    println!("  ✅ Preprocessing pipeline: {processed_chunks} chunks processed");

    Ok(())
}

Trait Implementations

impl Default for StreamTransformer

fn default() -> Self

Returns the “default value” for a type.