blob: b263c720639d8ff6e5481244ca00a836761572f1 (
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
|
(
(let mr_source (DataSource 'yt 'plato))
(let x (Read! world mr_source
(Key '('table (String 'Input)))
(Void) '()))
(let world (Left! x))
(let table1 (Right! x))
(let preMap (lambda '(item) (Just item)))
(let keyExtractor (lambda '(item) (Member item 'key)))
(let init (lambda '(key item) (FromString (Member item 'subkey) 'Uint32)))
(let update (lambda '(key item state) (+ state (FromString (Member item 'subkey) 'Uint32))))
(let finish (lambda '(key state) (block '(
(let s (Struct))
(let s (AddMember s 'key key))
(let s (AddMember s 'subkey (String '.)))
(let s (AddMember s 'value (Coalesce (Map state (lambda '(x) (ToString x))) (String '""))))
(return (Just s))
))))
(let table2 (CombineByKey table1 preMap keyExtractor init update finish))
(let table2 (Take table2 (Uint64 '2)))
(let res_sink (DataSink 'result))
(let world (Write! world res_sink
(Key)
table2 '()))
(let world (Commit! world res_sink))
(return world)
)
|
