blob: d4f0adca764dfae7cdb3d52494fb79f356acc5ee (
plain) (
blame)
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
|
(
#comment
(let res_sink (DataSink 'result))
(let mr_source (DataSource 'yt 'plato))
(let x (Read! world mr_source (Key '('table (String 'Input))) '('key 'subkey 'value) '()))
(let world (Left! x))
(let table1 (Right! x))
(let table1 (FlatMap table1 (lambda '(row) (block '(
(let res (Struct))
(let res (AddMember res '"key" '((Member row '"key") (Member row '"key"))))
(let res (AddMember res '"subkey" (Member row '"subkey")))
(let res (AddMember res '"value" (Member row 'value)))
(let res (AsList res))
(return res)
)))))
(let init (lambda '(x) (StrictFromString (Member x 'value) 'Uint32)))
(let init_distinct (lambda '(x) (StrictFromString x 'Uint32)))
(let update_sum_distinct (lambda '(x y) (+ (StrictFromString x 'Uint32) y)))
(let save (lambda '(x) x))
(let load (lambda '(x) x))
(let merge_sum (lambda '(x y) (+ x y)))
(let finish (lambda '(x) x))
# distinct process one column and requires data/data? type
(let sum (AggregationTraits (StructMemberType (ListItemType (TypeOf table1)) 'value) init_distinct update_sum_distinct save load merge_sum finish (Null)))
(let resAll (Aggregate table1 '() '('('distsum sum 'value))))
(let world (Write! world res_sink (Key) resAll '('('type))))
(let resKey (Aggregate table1 '('key) '('('distsum sum 'value))))
(let sortedOutput (Sort resKey '((Bool 'false) (Bool 'false)) (lambda '(x) (Member x 'key) )))
(let world (Write! world res_sink (Key) sortedOutput '('('type))))
(let resKey (Aggregate table1 '('key 'subkey) '('('distsum sum 'value))))
(let sortedOutput (Sort resKey '((Bool 'false) (Bool 'false) (Bool 'false)) (lambda '(x) '((Nth (Member x 'key) '0) (Nth (Member x 'key) '1) (Member x 'subkey)) )))
(let world (Write! world res_sink (Key) sortedOutput '('('type))))
(let world (Commit! world res_sink))
(return world)
)
|
