arrow2 0.7.1

Unofficial implementation of Apache Arrow spec in safe Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

use crossbeam_channel::unbounded;

use std::fs::File;
use std::sync::Arc;
use std::thread;
use std::time::SystemTime;

use arrow2::{array::Array, error::Result, io::parquet::read};

fn parallel_read(path: &str) -> Result<Vec<Box<dyn Array>>> {
    // prepare a channel to send serialized records from threads
    let (tx, rx) = unbounded();

    let mut file = File::open(path)?;
    let file_metadata = read::read_metadata(&mut file)?;
    let arrow_schema = Arc::new(read::get_schema(&file_metadata)?);

    let file_metadata = Arc::new(file_metadata);

    let start = SystemTime::now();
    // spawn a thread to produce `Vec<CompressedPage>` (IO bounded)
    let producer_metadata = file_metadata.clone();
    let child = thread::spawn(move || {
        for column in 0..producer_metadata.schema().num_columns() {
            for row_group in 0..producer_metadata.row_groups.len() {
                let start = SystemTime::now();
                let column_metadata = producer_metadata.row_groups[row_group].column(column);
                println!("produce start: {} {}", column, row_group);
                let pages = read::get_page_iterator(column_metadata, &mut file, None, vec![])
                    .unwrap()
                    .collect::<Vec<_>>();
                println!(
                    "produce end - {:?}: {} {}",
                    start.elapsed().unwrap(),
                    column,
                    row_group
                );
                tx.send((column, row_group, pages)).unwrap();
            }
        }
    });

    let mut children = Vec::new();
    // use 3 consumers of to decompress, decode and deserialize.
    for _ in 0..3 {
        let rx_consumer = rx.clone();
        let metadata_consumer = file_metadata.clone();
        let arrow_schema_consumer = arrow_schema.clone();
        let child = thread::spawn(move || {
            let (column, row_group, pages) = rx_consumer.recv().unwrap();
            let start = SystemTime::now();
            println!("consumer start - {} {}", column, row_group);
            let metadata = metadata_consumer.row_groups[row_group].column(column);
            let data_type = arrow_schema_consumer.fields()[column].data_type().clone();

            let mut pages = read::BasicDecompressor::new(pages.into_iter(), vec![]);

            let array = read::page_iter_to_array(&mut pages, metadata, data_type);
            println!(
                "consumer end - {:?}: {} {}",
                start.elapsed().unwrap(),
                column,
                row_group
            );
            array
        });
        children.push(child);
    }

    child.join().expect("child thread panicked");

    let arrays = children
        .into_iter()
        .map(|x| x.join().unwrap())
        .collect::<Result<Vec<_>>>()?;
    println!("Finished - {:?}", start.elapsed().unwrap());

    Ok(arrays)
}

fn main() -> Result<()> {
    use std::env;
    let args: Vec<String> = env::args().collect();
    let file_path = &args[1];

    let arrays = parallel_read(file_path)?;
    for array in arrays {
        println!("{}", array)
    }
    Ok(())
}