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
|
package main
import (
"context"
"errors"
"fmt"
"goflink/core"
"goflink/core/operator"
"goflink/runtime"
"goflink/state"
"goflink/transport"
)
// wordCounter counts how many times each key has been seen, in keyed state.
type wordCounter struct {
count core.ValueState[any]
}
func (w *wordCounter) Open(ctx context.Context) error {
be, ok := core.ExtractStateBackend(ctx)
if !ok {
return errors.New("no state backend in context")
}
s, err := be.GetValueState(ctx, "word-count")
if err != nil {
return err
}
w.count = s
return nil
}
func (w *wordCounter) Map(ctx context.Context, in string) (int, error) {
v, err := w.count.Value(ctx)
if err != nil {
return 0, err
}
n, _ := v.(int) // unseen key -> nil -> 0
n++
_, err = w.count.Update(ctx, n)
return n, err
}
func (w *wordCounter) Close(ctx context.Context) error { return nil }
func main() {
/*
Bây giờ bạn đã có đầy đủ đồ chơi: MemoryStateBackend, KeyByOperator, StatefulMapOperator.
Trong file main.go, bạn hãy thiết lập một Pipeline có dạng: Source ➡️ KeyBy (Trích xuất Tên) ➡️ StatefulMap (Đếm số lần Tên xuất hiện) ➡️ Sink
Yêu cầu Logic của hàm StatefulMap:
Trong hàm Open, bạn hãy khởi tạo một ValueState[int] tên là "word-count".
Trong hàm xử lý (User Func), bạn hãy đọc giá trị cũ lên, cộng thêm 1, lưu lại vào State, và trả về con số mới đó.
Hãy chạy thử xem nếu bạn truyền vào ["Sang", "Tran", "Sang", "Sang"], nó có in ra 1, 1, 2, 3 không nhé. Nếu thành công, bạn đã tự tay chế tạo xong một Streaming Engine Stateful đích thực!
*/
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// ponytail: one backend for the whole job. Each stateful task needs its own
// once tasks run on separate key-group shards.
ctx = core.InjectStateBackend(ctx, state.NewMemoryStateBackend(core.DefaultMaxParallelism))
source := make(chan core.StreamRecord[string])
mid := make(chan core.StreamRecord[string])
sink := make(chan core.StreamRecord[int])
// keyBy(name) -> count occurrences per name
keyed := runtime.RunTask(ctx,
transport.NewLocalReceiveChannel(source),
transport.NewLocalEmitChannel(mid),
operator.NewKeyByOperator(func(in string) string { return in }),
)
counted := runtime.RunTask(ctx,
transport.NewLocalReceiveChannel(mid),
transport.NewLocalEmitChannel(sink),
operator.NewStatefulMapOperator[string, int](&wordCounter{}),
)
go func() {
defer close(source)
for _, name := range []string{"Sang", "Tran", "Sang", "Sang"} {
source <- core.NewDataRecord("", name, 0)
}
}()
for record := range sink {
fmt.Println(record.Key, record.Value)
}
for _, errCh := range []<-chan error{keyed, counted} {
if err := <-errCh; err != nil {
fmt.Println("task failed:", err)
}
}
}
|