dataflow 0.4.0

Dataflow is a data processing library, primarily for machine learning.
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

use std::{collections::BTreeMap, marker::PhantomData};

use crate::pipeline::Node;

/// Implements the MapReduce operation as seen here: https://research.google/pubs/pub62/
pub struct MapReduce<I, K, V, O, Map: Fn(I) -> Vec<(K, V)>, Reduce: Fn((K, Vec<V>)) -> Vec<O>> {
    map: Map,
    reduce: Reduce,
    _phantom: PhantomData<(I, K, V, O)>,
}

impl<I, K, V, O, Map: Fn(I) -> Vec<(K, V)> + Clone, Reduce: Fn((K, Vec<V>)) -> Vec<O> + Clone> Clone
    for MapReduce<I, K, V, O, Map, Reduce>
{
    fn clone(&self) -> Self {
        Self {
            map: self.map.clone(),
            reduce: self.reduce.clone(),
            _phantom: self._phantom,
        }
    }
}

impl<I, K: Ord, V, O, Map: Fn(I) -> Vec<(K, V)>, Reduce: Fn((K, Vec<V>)) -> Vec<O>>
    MapReduce<I, K, V, O, Map, Reduce>
{
    pub fn new(map: Map, reduce: Reduce) -> Self {
        Self {
            map,
            reduce,
            _phantom: PhantomData::default(),
        }
    }
}

impl<I, K: Ord, V, O, Map: Fn(I) -> Vec<(K, V)>, Reduce: Fn((K, Vec<V>)) -> Vec<O>> Node<Vec<I>>
    for MapReduce<I, K, V, O, Map, Reduce>
{
    type Output = Vec<O>;

    fn process(&mut self, input: Vec<I>) -> Self::Output {
        group(input.into_iter().flat_map(&self.map))
            .into_iter()
            .flat_map(&self.reduce)
            .collect()
    }
}

fn group<A, B, I>(v: I) -> BTreeMap<A, Vec<B>>
where
    A: Ord,
    I: IntoIterator<Item = (A, B)>,
{
    let mut result = BTreeMap::<A, Vec<B>>::new();
    for (a, b) in v {
        result.entry(a).or_default().push(b);
    }
    result
}