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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
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 string] struct {
totalTransaction int
average int
}
func (t *TransactionSummary[IN, OUT]) Apply(ctx context.Context, key string, window core.Window, in []Transaction, out operator.Emitter[string]) error {
totalAmount := 0
for _, v := range in {
totalAmount += v.Amount
}
msg := fmt.Sprintf("Khách: %s | Số đơn: %d | Tổng tiền: %d$", key, len(in), totalAmount)
return out.Emit(ctx, core.StreamRecord[string]{
Value: msg,
Type: core.RecordData,
Timestamp: window.MaxTimestamp(),
})}
func main() {
// init
ctx := context.Background()
// memory backend
// backend := state.NewMemoryStateBackend(1)
// pebble backend
backend, err := state.NewPebbleStateBackend("./data_pebble", 1)
if err != nil {
panic(err)
}
defer backend.Close()
ctx = core.InjectStateBackend(ctx, backend)
txList := []Transaction{
{
User: "Sang",
Amount: 1,
Timestamp: 1000,
},
{
User: "Sang",
Amount: 4,
Timestamp: 1500,
},
{
User: "Sang",
Amount: 10,
Timestamp: 2000,
},
{
User: "Sang",
Amount: 12,
Timestamp: 2500,
},
{
User: "Sang",
Amount: 10,
Timestamp: 3000,
},
}
ch1 := make(chan core.StreamRecord[Transaction], len(txList))
ch2 := make(chan core.StreamRecord[Transaction], len(txList))
ch3 := make(chan core.StreamRecord[Transaction], len(txList))
ch4 := make(chan core.StreamRecord[string], len(txList))
wo := operator.NewWatermarkGeneratorOperator[Transaction](0, func(in Transaction) int64 {
return in.Timestamp
})
ko := operator.NewKeyByOperator[Transaction](func(in Transaction) string {
return in.User
})
wop := operator.NewWindowOperator[Transaction, string](core.NewTumblingWindows[Transaction](1000), &TransactionSummary[Transaction, string]{})
var wg sync.WaitGroup
runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch1), transport.NewLocalEmitChannel(ch2), wo)
runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch2), transport.NewLocalEmitChannel(ch3), ko)
runtime.RunTask(ctx, transport.NewLocalReceiveChannel(ch3), transport.NewLocalEmitChannel(ch4), wop)
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 ch4 {
if v.Type == core.RecordData {
fmt.Println("Total Amount after: ", v.Value)
}
}
})
wg.Wait()
}
|
