사례 (II) - KisFlow-Golang 스트림 실시간 컴퓨팅 - 흐름 병렬 작업
- PHPz원래의
- 2024-07-16 17:28:10792검색
Github: https://github.com/aceld/kis-flow
문서: https://github.com/aceld/kis-flow/wiki
1부-개요
Part2.1-프로젝트 구성 / 기본 모듈
Part2.2-프로젝트 구성 / 기본 모듈
Part3-데이터 스트림
Part4-기능 스케줄링
5부-커넥터
Part6-구성 가져오기 및 내보내기
Part7-KisFlow 액션
Part8-캐시/매개변수 데이터 캐싱 및 데이터 매개변수
Part9-흐름의 다중 복사본
Part10-프로메테우스 지표 통계
Part11 - Reflection을 기반으로 한 FaaS 매개변수 유형의 적응적 등록
사례1-빠른 시작
Case2-Flow 병렬운전
사례3 - 다중 고루틴에 KisFlow 적용
Message Queue(MQ) 애플리케이션의 Case4-KisFlow
KisFlow 소스 다운로드
$go get github.com/aceld/kis-flow
KisFlow 개발자 문서
소스 코드 예
https://github.com/aceld/kis-flow-usage/tree/main/8-connector
KisFlow는 커넥터를 통해 두 흐름의 결합을 이룰 수 있습니다
이 소개에서는 다음 두 가지 흐름을 조합하여 커넥터의 인터페이스와 사용법을 다룹니다.
데이터 흐름 다이어그램
사례소개
학생에게 다음 네 가지 특성이 있다고 가정합니다.
Student ID: stu_id Credit 1: score_1 Credit 2: score_2 Credit 3: score_3
Flow1: CalStuAvgScore-1-2를 정의하여 학생의 평균 점수인 1학점(score_1) 및 2학점(score_2)(avg_score_1_2)을 계산합니다.
Flow2: CalStuAvgScore-3을 정의하여 학생의 평균 점수인 Credit 3(score_3)과 avg_score_1_2(1학점, 2학점, 3학점의 평균)를 계산합니다. 1학점과 2학점의 평균은 Flow1에서 제공됩니다.
흐름1
Flow1은 4가지 기능으로 구성됩니다.
StuId의 유효성을 검증하는 V(VerifyStu 기능)
C(함수: AvgStuScore12) 1학점과 2학점의 평균 점수를 계산합니다
S(함수: SaveScoreAvg12): avg_score_1_2를 Redis에 저장
E(함수: PrintStuAvgScore)는 1학점과 2학점의 평균 점수를 인쇄합니다.
흐름2
Flow2는 4가지 기능으로 구성됩니다.
StuId의 유효성을 검증하는 V(VerifyStu 기능)
L(함수: LoadScoreAvg12) Flow1에서 계산한 현재 학생의 1학점과 2학점 평균 점수(avg_score_1_2)를 읽어옵니다
C(함수: AvgStuScore3)는 3학점의 평균 점수와 1학점과 2학점의 평균 점수를 계산합니다
E(함수: PrintStuAvgScore)는 1학점, 2학점, 3학점의 평균 점수를 인쇄합니다.
conf/func/func-AvgStuScore-3.yml
kistype: func
fname: AvgStuScore3
fmode: Calculate
source:
name: SourceStuScore
must:
- stu_id
conf/func/func-LoadScoreAvg-1-2.yml
kistype: func
fname: LoadScoreAvg12
fmode: Load
source:
name: SourceStuScore
must:
- stu_id
option:
cname: Score12Cache
기본 데이터 프로토콜
stu_proto.go
package main
type StuScore1_2 struct {
StuId int `json:"stu_id"`
Score1 int `json:"score_1"`
Score2 int `json:"score_2"`
}
type StuScoreAvg struct {
StuId int `json:"stu_id"`
AvgScore float64 `json:"avg_score"`
}
type StuScore3 struct {
StuId int `json:"stu_id"`
AvgScore12 float64 `json:"avg_score_1_2"` // score_1, score_2 avg
Score3 int `json:"score_3"`
}
커넥터 초기화
이 프로젝트에서 정의한 커넥터인 Score12Cache는 Redis와 관련된 링크 리소스입니다. 이 커넥터는 KisFlow 시작 시 연결을 설정하기 위한 초기화 방법이 필요합니다.
conn_init.go
package main
import (
"context"
"fmt"
"github.com/aceld/kis-flow/kis"
"github.com/aceld/kis-flow/log"
"github.com/go-redis/redis/v8"
)
// type ConnInit func(conn Connector) error
func InitScore12Cache(connector kis.Connector) error {
fmt.Println("===> Call Connector InitScore12Cache")
// init Redis Conn Client
rdb := redis.NewClient(&redis.Options{
Addr: connector.GetConfig().AddrString, // Redis-Server address
Password: "", // password
DB: 0, // select db
})
// Ping test
pong, err := rdb.Ping(context.Background()).Result()
if err != nil {
log.Logger().ErrorF("Failed to connect to Redis: %v", err)
return err
}
fmt.Println("Connected to Redis:", pong)
// set rdb to connector
connector.SetMetaData("rdb", rdb)
return nil
}
여기서 성공적으로 연결된 Redis 인스턴스는 커넥터의 캐시 변수 "rdb"에 저장됩니다.
// set rdb to connector
connector.SetMetaData("rdb", rdb)
FaaS 구현
기능(V): 검증Stu
faas_stu_verify.go
package main
import (
"context"
"fmt"
"github.com/aceld/kis-flow/kis"
"github.com/aceld/kis-flow/serialize"
)
type VerifyStuIn struct {
serialize.DefaultSerialize
StuId int `json:"stu_id"`
}
func VerifyStu(ctx context.Context, flow kis.Flow, rows []*VerifyStuIn) error {
fmt.Printf("->Call Func VerifyStu\n")
for _, stu := range rows {
// Filter out invalid data
if stu.StuId < 0 || stu.StuId > 999 {
// Terminate the current Flow process, subsequent functions of the current Flow will not be executed
return flow.Next(kis.ActionAbort)
}
}
return flow.Next(kis.ActionDataReuse)
}
VerifyStu()는 데이터의 유효성을 검사하는 데 사용됩니다. 데이터가 요구 사항을 충족하지 않으면 현재 데이터 흐름이 종료됩니다. 마지막으로 데이터는 재사용되어 flow.Next(kis.ActionDataReuse)를 통해 다음 레이어로 전달됩니다.
함수(C): AvgStuScore12
faas_avg_score_1_2.go
package main
import (
"context"
"fmt"
"github.com/aceld/kis-flow/kis"
"github.com/aceld/kis-flow/serialize"
)
type AvgStuScoreIn_1_2 struct {
serialize.DefaultSerialize
StuScore1_2
}
type AvgStuScoreOut_1_2 struct {
serialize.DefaultSerialize
StuScoreAvg
}
func AvgStuScore12(ctx context.Context, flow kis.Flow, rows []*AvgStuScoreIn_1_2) error {
fmt.Printf("->Call Func AvgStuScore12\n")
for _, row := range rows {
out := AvgStuScoreOut_1_2{
StuScoreAvg: StuScoreAvg{
StuId: row.StuId,
AvgScore: float64(row.Score1+row.Score2) / 2,
},
}
// Submit result data
_ = flow.CommitRow(out)
}
return flow.Next()
}
AvgStuScore12()는 점수_1과 점수_2의 평균 점수를 계산하여 avg_score를 얻습니다.
함수(S): SaveScoreAvg12
faas_save_score_avg_1_2.go
package main
import (
"context"
"fmt"
"github.com/aceld/kis-flow/kis"
"github.com/aceld/kis-flow/serialize"
"github.com/go-redis/redis/v8"
"strconv"
)
type SaveStuScoreIn struct {
serialize.DefaultSerialize
StuScoreAvg
}
func BatchSetStuScores(ctx context.Context, conn kis.Connector, rows []*SaveStuScoreIn) error {
var rdb *redis.Client
// Get Redis Client
rdb = conn.GetMetaData("rdb").(*redis.Client)
// Set data to redis
pipe := rdb.Pipeline()
for _, score := range rows {
// make key
key := conn.GetConfig().Key + strconv.Itoa(score.StuId)
pipe.HMSet(context.Background(), key, map[string]interface{}{
"avg_score": score.AvgScore,
})
}
_, err := pipe.Exec(ctx)
if err != nil {
return err
}
return nil
}
func SaveScoreAvg12(ctx context.Context, flow kis.Flow, rows []*SaveStuScoreIn) error {
fmt.Printf("->Call Func SaveScoreAvg12\n")
conn, err := flow.GetConnector()
if err != nil {
fmt.Printf("SaveScoreAvg12(): GetConnector err = %s\n", err.Error())
return err
}
if BatchSetStuScores(ctx, conn, rows) != nil {
fmt.Printf("SaveScoreAvg12(): BatchSetStuScores err = %s\n", err.Error())
return err
}
return flow.Next(kis.ActionDataReuse)
}
SaveScoreAvg12()는 커넥터에 구성된 키를 사용하여 바인딩된 커넥터를 통해 Redis에 데이터를 저장합니다. 마지막으로 소스 데이터가 다음 기능으로 투명하게 전달됩니다.
함수(E): PrintStuAvgScore
faas_stu_score_avg_print.go
package main
import (
"context"
"fmt"
"github.com/aceld/kis-flow/kis"
"github.com/aceld/kis-flow/serialize"
)
type PrintStuAvgScoreIn struct {
serialize.DefaultSerialize
StuId int `json:"stu_id"`
AvgScore float64 `json:"avg_score"`
}
func PrintStuAvgScore(ctx context.Context, flow kis.Flow, rows []*PrintStuAvgScoreIn) error {
fmt.Printf("->Call Func PrintStuAvgScore, in Flow[%s]\n", flow.GetName())
for _, row := range rows {
fmt.Printf("stuid: [%+v], avg score: [%+v]\n", row.StuId, row.AvgScore)
}
return flow.Next()
}
PrintStuAvgScore()는 현재 학생의 평균 점수를 인쇄합니다.
함수(L): LoadScoreAvg12
faas_load_score_avg_1_2.go
package main
import (
"context"
"fmt"
"github.com/aceld/kis-flow/kis"
"github.com/aceld/kis-flow/serialize"
"github.com/go-redis/redis/v8"
"strconv"
)
type LoadStuScoreIn struct {
serialize.DefaultSerialize
StuScore3
}
type LoadStuScoreOut struct {
serialize.DefaultSerialize
StuScore3
}
func GetStuScoresByStuId(ctx context.Context, conn kis.Connector, stuId int) (float64, error) {
var rdb *redis.Client
// Get Redis Client
rdb = conn.GetMetaData("rdb").(*redis.Client)
// make key
key := conn.GetConfig().Key + strconv.Itoa(stuId)
// get data from redis
result, err := rdb.HGetAll(ctx, key).Result()
if err != nil {
return 0, err
}
// get value
avgScoreStr, ok := result["avg_score"]
if !ok {
return 0, fmt.Errorf("avg_score not found for stuId: %d", stuId)
}
// parse to float64
avgScore, err := strconv.ParseFloat(avgScoreStr, 64)
if err != nil {
return 0, err
}
return avgScore, nil
}
func LoadScoreAvg12(ctx context.Context, flow kis.Flow, rows []*LoadStuScoreIn) error {
fmt.Printf("->Call Func LoadScoreAvg12\n")
conn, err := flow.GetConnector()
if err != nil {
fmt.Printf("LoadScoreAvg12(): GetConnector err = %s\n", err.Error())
return err
}
for _, row := range rows {
stuScoreAvg1_2, err := GetStuScoresByStuId(ctx, conn, row.StuId)
if err != nil {
fmt.Printf("LoadScoreAvg12(): GetStuScoresByStuId err = %s\n", err.Error())
return err
}
out := LoadStuScoreOut{
StuScore3: StuScore3{
StuId: row.StuId,
Score3: row.Score3,
AvgScore12: stuScoreAvg1_2, // avg score of score1 and score2 (load from redis)
},
}
// commit result
_ = flow.CommitRow(out)
}
return flow.Next()
}
LoadScoreAvg12() reads the average score of score_1 and score_2 from Redis through the linked resource Redis of the bound Connector using the key configured in the Connector. It then sends the source data from upstream, along with the newly read average score of score1 and score2, to the next layer.
Function(C): AvgStuScore3
faas_stu_score_avg_3.go
package main
import (
"context"
"fmt"
"github.com/aceld/kis-flow/kis"
"github.com/aceld/kis-flow/serialize"
)
type AvgStuScore3In struct {
serialize.DefaultSerialize
StuScore3
}
type AvgStuScore3Out struct {
serialize.DefaultSerialize
StuScoreAvg
}
func AvgStuScore3(ctx context.Context, flow kis.Flow, rows []*AvgStuScore3In) error {
fmt.Printf("->Call Func AvgStuScore3\n")
for _, row := range rows {
out := AvgStuScore3Out{
StuScoreAvg: StuScoreAvg{
StuId: row.StuId,
AvgScore: (float64(row.Score3) + row.AvgScore12*2) / 3,
},
}
// Submit result data
_ = flow.CommitRow(out)
}
return flow.Next()
}
AvgStuScore3() recalculates the average score of three scores by adding score_3 and the average score of score_1 and score_2, resulting in the final average score avg_score.
Register FaaS & CaaSInit/CaaS (Register Function/Connector)
main.go
func init() {
// Register functions
kis.Pool().FaaS("VerifyStu", VerifyStu)
kis.Pool().FaaS("AvgStuScore12", AvgStuScore12)
kis.Pool().FaaS("SaveScoreAvg12", SaveScoreAvg12)
kis.Pool().FaaS("PrintStuAvgScore", PrintStuAvgScore)
kis.Pool().FaaS("LoadScoreAvg12", LoadScoreAvg12)
kis.Pool().FaaS("AvgStuScore3", AvgStuScore3)
// Register connectors
kis.Pool().CaaSInit("Score12Cache", InitScore12Cache)
}
Main Process
main.go
package main
import (
"context"
"github.com/aceld/kis-flow/file"
"github.com/aceld/kis-flow/kis"
"sync"
)
func RunFlowCalStuAvgScore12(ctx context.Context, flow kis.Flow) error {
// Commit data
_ = flow.CommitRow(`{"stu_id":101, "score_1":100, "score_2":90}`)
_ = flow.CommitRow(`{"stu_id":102, "score_1":100, "score_2":80}`)
// Run the flow
if err := flow.Run(ctx); err != nil {
return err
}
return nil
}
func RunFlowCalStuAvgScore3(ctx context.Context, flow kis.Flow) error {
// Commit data
_ = flow.CommitRow(`{"stu_id":101, "score_3": 80}`)
_ = flow.CommitRow(`{"stu_id":102, "score_3": 70}`)
// Run the flow
if err := flow.Run(ctx); err != nil {
return err
}
return nil
}
func main() {
ctx := context.Background()
// Load Configuration from file
if err := file.ConfigImportYaml("conf/"); err != nil {
panic(err)
}
var wg sync.WaitGroup
wg.Add(2)
go func() {
// Run flow1 concurrently
defer wg.Done()
flow1 := kis.Pool().GetFlow("CalStuAvgScore12")
if flow1 == nil {
panic("flow1 is nil")
}
if err := RunFlowCalStuAvgScore12(ctx, flow1); err != nil {
panic(err)
}
}()
go func() {
// Run flow2 concurrently
defer wg.Done()
flow2 := kis.Pool().GetFlow("CalStuAvgScore3")
if flow2 == nil {
panic("flow2 is nil")
}
if err := RunFlowCalStuAvgScore3(ctx, flow2); err != nil {
panic(err)
}
}()
wg.Wait()
return
}
Two Goroutines are launched concurrently to execute Flow1 and Flow2, calculating the final average scores for student 101 and student 102.
Execution Results
===> Call Connector InitScore12Cache Connected to Redis: PONG Add FlowRouter FlowName=CalStuAvgScore12 ===> Call Connector InitScore12Cache Connected to Redis: PONG Add FlowRouter FlowName=CalStuAvgScore3 ->Call Func VerifyStu ->Call Func VerifyStu ->Call Func AvgStuScore12 ->Call Func LoadScoreAvg12 ->Call Func SaveScoreAvg12 ->Call Func PrintStuAvgScore, in Flow[CalStuAvgScore12] stuid: [101], avg score: [95] stuid: [102], avg score: [90] ->Call Func AvgStuScore3 ->Call Func PrintStuAvgScore, in Flow[CalStuAvgScore3] stuid: [101], avg score: [90] stuid: [102], avg score: [83.33333333333333]
In Flow[CalStuAvgScore3], we observe the final computed average scores for scores 1, 2, and 3.
Author: Aceld
GitHub: https://github.com/aceld
KisFlow Open Source Project Address: https://github.com/aceld/kis-flow
Document: https://github.com/aceld/kis-flow/wiki
Part1-OverView
Part2.1-Project Construction / Basic Modules
Part2.2-Project Construction / Basic Modules
Part3-Data Stream
Part4-Function Scheduling
Part5-Connector
Part6-Configuration Import and Export
Part7-KisFlow Action
Part8-Cache/Params Data Caching and Data Parameters
Part9-Multiple Copies of Flow
Part10-Prometheus Metrics Statistics
Part11-Adaptive Registration of FaaS Parameter Types Based on Reflection
Case1-Quick Start
Case2-Flow Parallel Operation
Case3-Application of KisFlow in Multi-Goroutines
Case4-KisFlow in Message Queue (MQ) Applications
위 내용은 사례 (II) - KisFlow-Golang 스트림 실시간 컴퓨팅 - 흐름 병렬 작업의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!