Struct StreamBatchIterator 

pub struct StreamBatchIterator { /* private fields */ }

Iterator for scanning Redis Streams in batches as Arrow RecordBatches.

This iterator fetches stream keys matching a pattern and retrieves their entries, converting them to Arrow RecordBatches for use with Polars. Each entry becomes a row in the output.

Supports two modes:

1. Single stream: Provide a specific key and iterate through entries
2. Multi-stream: Provide a pattern to SCAN and fetch entries from each

Example

use polars_redis::{StreamBatchIterator, StreamSchema, BatchConfig};

let schema = StreamSchema::new(vec![
    ("user_id".to_string(), DataType::Utf8),
    ("action".to_string(), DataType::Utf8),
]).with_key(true).with_entry_id(true);

let config = BatchConfig::new("events:*").with_batch_size(1000);

let mut iterator = StreamBatchIterator::new(url, schema, config)?;

while let Some(batch) = iterator.next_batch()? {
    println!("Got {} entries", batch.num_rows());
}

Output Schema

• _key (optional): The Redis stream key
• _entry_id (optional): The stream entry ID (e.g., “1234567890-0”)
• _timestamp (optional): Entry timestamp extracted from ID (Int64 ms)
• User-defined fields from the stream entries

Implementations

impl StreamBatchIterator

pub fn new(url: &str, schema: StreamSchema, config: BatchConfig) -> Result<Self>

Create a new StreamBatchIterator.

Arguments

• url - Redis connection URL
• schema - Schema configuration for the stream
• config - Batch configuration (pattern, batch_size, etc.)

pub fn with_start_id(self, id: &str) -> Self

Set the start ID for XRANGE.

pub fn with_end_id(self, id: &str) -> Self

Set the end ID for XRANGE.

pub fn with_count_per_stream(self, count: usize) -> Self

Set the maximum entries to fetch per stream.

pub fn next_batch(&mut self) -> Result<Option<RecordBatch>>

Get the next batch of data as a RecordBatch.

pub fn is_done(&self) -> bool

Check if iteration is complete.

pub fn rows_yielded(&self) -> usize

Get the number of rows (entries) yielded so far.