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
92
93
94
95
96
|
package main
import (
"context"
"fmt"
"strconv"
"sync"
"flux/core"
"flux/core/operator"
"flux/runtime"
"flux/state"
"flux/transport"
)
type Transaction struct {
User string
Amount int
Timestamp int64
}
type TransactionSummary[IN Transaction, OUT Transaction] struct {
totalTransaction int
totalAmount int
average int
}
func (t *TransactionSummary[IN, OUT]) Apply(ctx context.Context, key string, window core.Window, in []Transaction, out transport.Emitter[Transaction]) error {
for _, v := range in {
t.totalAmount += v.Amount
out.Emit(ctx, core.StreamRecord[Transaction]{
Value: v,
Type: core.RecordData,
Timestamp: v.Timestamp,
})
}
return nil
}
func main() {
// init
ctx := context.Background()
backend := state.NewMemoryStateBackend(1)
ctx = core.InjectStateBackend(ctx, backend)
txList := []Transaction{
{
User: "Sang",
Amount: 1,
Timestamp: 1000,
},
{
User: "Sang",
Amount: 4,
Timestamp: 1200,
},
{
User: "Sang",
Amount: 10,
Timestamp: 1600,
},
}
ch1 := make(chan core.StreamRecord[Transaction], len(txList))
ch2 := make(chan core.StreamRecord[Transaction], len(txList))
wo := operator.NewWatermarkGeneratorOperator[Transaction](0, func(in Transaction) int64 {
return in.Timestamp
})
var wg sync.WaitGroup
runtime.RunTask[Transaction, Transaction](ctx, transport.NewLocalReceiveChannel(ch1), transport.NewLocalEmitChannel(ch2), wo)
wg.Go(func() {
for i, tx := range txList {
ch1 <- core.NewDataRecord(strconv.Itoa(i), tx, tx.Timestamp)
}
close(ch1)
})
wg.Go(func() {
for v := range ch2 {
if v.Type == core.RecordData {
fmt.Println("data: ", v.Value)
} else {
fmt.Println("watermark: ", v.Timestamp)
}
}
})
wg.Wait()
}
|
