Struct StreamWriter 

pub struct StreamWriter<W> {
    writer: W,
    write_options: IpcWriteOptions,
    finished: bool,
    dictionary_tracker: DictionaryTracker,
    data_gen: IpcDataGenerator,
    compression_context: CompressionContext,
}

Arrow Stream Writer

Writes Arrow RecordBatches to bytes using the IPC Streaming Format.

See Also

• FileWriter for writing IPC Files

Example - Basic usage

let batch = record_batch!(("a", Int32, [1, 2, 3])).unwrap();
// create a new writer, the schema must be known in advance
let mut writer = StreamWriter::try_new(&mut stream, &batch.schema()).unwrap();
// write each batch to the underlying stream
writer.write(&batch).unwrap();
// When all batches are written, call finish to flush all buffers
writer.finish().unwrap();

Example - Efficient delta dictionaries

let schema = Arc::new(Schema::new(vec![Field::new(
   "col1",
   DataType::Dictionary(Box::from(DataType::Int32), Box::from(DataType::Utf8)),
   true,
)]));

let mut builder = StringDictionaryBuilder::<arrow_array::types::Int32Type>::new();

// `finish_preserve_values` will keep the dictionary values along with their
// key assignments so that they can be re-used in the next batch.
builder.append("a").unwrap();
builder.append("b").unwrap();
let array1 = builder.finish_preserve_values();
let batch1 = RecordBatch::try_new(schema.clone(), vec![Arc::new(array1) as ArrayRef]).unwrap();

// In this batch, 'a' will have the same dictionary key as 'a' in the previous batch,
// and 'd' will take the next available key.
builder.append("a").unwrap();
builder.append("d").unwrap();
let array2 = builder.finish_preserve_values();
let batch2 = RecordBatch::try_new(schema.clone(), vec![Arc::new(array2) as ArrayRef]).unwrap();

let mut stream = vec![];
// You must set `.with_dictionary_handling(DictionaryHandling::Delta)` to
// enable delta dictionaries in the writer
let options = IpcWriteOptions::default().with_dictionary_handling(DictionaryHandling::Delta);
let mut writer = StreamWriter::try_new(&mut stream, &schema).unwrap();

// When writing the first batch, a dictionary message with 'a' and 'b' will be written
// prior to the record batch.
writer.write(&batch1).unwrap();
// With the second batch only a delta dictionary with 'd' will be written
// prior to the record batch. This is only possible with `finish_preserve_values`.
// Without it, 'a' and 'd' in this batch would have different keys than the
// first batch and so we'd have to send a replacement dictionary with new keys
// for both.
writer.write(&batch2).unwrap();
writer.finish().unwrap();

Fields

writer: W

write_options: IpcWriteOptions

finished: bool

dictionary_tracker: DictionaryTracker

data_gen: IpcDataGenerator

compression_context: CompressionContext

Implementations

impl<W> StreamWriter<BufWriter<W>>

where W: Write,

pub fn try_new_buffered( writer: W, schema: &Schema, ) -> Result<StreamWriter<BufWriter<W>>, ArrowError>

Try to create a new stream writer with the writer wrapped in a BufWriter.

See StreamWriter::try_new for an unbuffered version.

impl<W> StreamWriter<W>

where W: Write,

pub fn try_new( writer: W, schema: &Schema, ) -> Result<StreamWriter<W>, ArrowError>

Try to create a new writer, with the schema written as part of the header.

Note that there is no internal buffering. See also StreamWriter::try_new_buffered.

Errors

An ‘Err’ may be returned if writing the header to the writer fails.

pub fn try_new_with_options( writer: W, schema: &Schema, write_options: IpcWriteOptions, ) -> Result<StreamWriter<W>, ArrowError>

Try to create a new writer with IpcWriteOptions.

Errors

An ‘Err’ may be returned if writing the header to the writer fails.

pub fn write(&mut self, batch: &RecordBatch) -> Result<(), ArrowError>

Write a record batch to the stream

pub fn finish(&mut self) -> Result<(), ArrowError>

Write continuation bytes, and mark the stream as done

pub fn get_ref(&self) -> &W

Gets a reference to the underlying writer.

pub fn get_mut(&mut self) -> &mut W

Gets a mutable reference to the underlying writer.

It is inadvisable to directly write to the underlying writer.

pub fn flush(&mut self) -> Result<(), ArrowError>

Flush the underlying writer.

Both the BufWriter and the underlying writer are flushed.

pub fn into_inner(self) -> Result<W, ArrowError>

Unwraps the the underlying writer.

The writer is flushed and the StreamWriter is finished before returning.

Errors

An ‘Err’ may be returned if an error occurs while finishing the StreamWriter or while flushing the writer.

Example

// The result we expect from an empty schema
let expected = vec![
    255, 255, 255, 255,  48,   0,   0,   0,
     16,   0,   0,   0,   0,   0,  10,   0,
     12,   0,  10,   0,   9,   0,   4,   0,
     10,   0,   0,   0,  16,   0,   0,   0,
      0,   1,   4,   0,   8,   0,   8,   0,
      0,   0,   4,   0,   8,   0,   0,   0,
      4,   0,   0,   0,   0,   0,   0,   0,
    255, 255, 255, 255,   0,   0,   0,   0
];

let schema = Schema::empty();
let buffer: Vec<u8> = Vec::new();
let options = IpcWriteOptions::try_new(8, false, MetadataVersion::V5)?;
let stream_writer = StreamWriter::try_new_with_options(buffer, &schema, options)?;

assert_eq!(stream_writer.into_inner()?, expected);

Trait Implementations

impl<W> RecordBatchWriter for StreamWriter<W>

where W: Write,

fn write(&mut self, batch: &RecordBatch) -> Result<(), ArrowError>

Write a single batch to the writer.

fn close(self) -> Result<(), ArrowError>

Write footer or termination data, then mark the writer as done.