- 微信
- 微博
  
  分享文章到微博
- 复制链接
  
  复制链接到剪贴板

Golang框架实战-KisFlow流式计算框架(1)-概述

刘丹冰Aceld 发表于 2024/04/23 11:54:02 2024/04/23

【摘要】一些大型toB企业级的项目，需要大量的业务数据，多数的数据需要流式实时计算的能力，但是很多公司还不足以承担一个数仓类似，Flink + Hadoop/HBase 等等。但是业务数据的实时计算需求依然存在，所以大多数的企业依然会让业务工程师来消化这些业务数据计算的工作。而这样只能直接查询业务数据库，这样会间接影响的业务能力，或定时任务/脚本来做定时计算，这些都不是好的办法。

kis-flow源代码： https://github.com/aceld/kis-flow

1.1 为什么需要KisFlow

一些大型toB企业级的项目，需要大量的业务数据，多数的数据需要流式实时计算的能力，但是很多公司还不足以承担一个数仓类似，Flink + Hadoop/HBase 等等。但是业务数据的实时计算需求依然存在，所以大多数的企业依然会让业务工程师来消化这些业务数据计算的工作。

而这样只能直接查询业务数据库，这样会间接影响的业务能力，或定时任务/脚本来做定时计算，这些都不是好的办法。本人亲身经历过一个大规模的系统，多达上千个需要计算的业务数据字段，而早期因为没有规划好，最后导致存在1000+的脚本在定时跑，最后导致了脚本之间对数据的影响，数据始终无法准确，导致业务数据经常性的报数据问题错误。如下面一个场景：某个业务计算字段的值，正确为100，错误为99，但是由于历史代码的臃肿，会有多个计算脚本对其值做修复补丁计算，会有各个脚本相互冲突，在一定的时间间隔内会存在数据值抖动，可能最终一个补丁修复正确，但是这种情况就会存在一定时间范围内业务数据不正确，最终却奇迹正确的情况，很让用户苦恼。

KisFlow就是为了解决当企业不具备数仓平台的计算能力，又依然存在大量数据实时计算的场景，让业务工程师可以投入到数据流式计算的业务中来，并且可以复用常用和通用的计算逻辑。

1.2 KisFlow实要支持的能力

流式计算

1、分布式批量消费能力（基于上游ODS消费配置：如Binlog、Kafka等）
2、Stateful Function能力，基于有状态的流式计算节点拼接，流式计算横纵向扩展。
3、数据流监控及修复能力，消费服务监控。
4、多流拼接及第三方中间件存储插件化。

分布式任务调度

5、分布式定时任务调度、日志监控、任务调度状态。
6、可视化调度平台。

1.3 KisFlow系统定位

KisFlow为业务上游计算层，上层接数仓/其他业务方ODS层、下游接本业务存储数据中心。

1.4 KisFlow整体架构图

层级	层级说明	包括子模块
流式计算层	为KisFlow上游计算层，直接对接业务存储及数仓ODS层，如上游可以为Mysql Binlog、日志、接口数据等，为被动消费模式，提供KisFlow实时计算能力。	KisFlow：分布式批量消费者,一个KisFlow是由多个KisFunction组合。 KisConnectors：计算数据流流中间状态持久存储及连接器。 KisFunctions：支持算子表达式拼接，Connectors集成、策略配置、Stateful Function模式、Slink流式拼接等。 KisConfig：KisFunction的绑定的流处理策略，可以绑定ReSource让Function具有固定的独立流处理能力。 KisSource：对接ODS的数据源
任务调度层	定时任务调度及执行器业务逻辑，包括任务调度平台、执行器管理、调度日志及用户管理等。提供KisFlow的定时任务、统计、聚合运算等调度计算能力。	任务调度平台可视化：包括任务的运行报表、调度报表、成功比例、任务管理、配置管理、GLUE IDE等可视化管理平台。执行器管理 KisJobs：Golang SDK及计算自定义业务逻辑、执行器的自动注册、任务触发、终止及摘除等。执行器场景KisScenes：根据业务划分的逻辑任务集合。调度日志及用户管理：任务调度日志收集、调度详细、调度流程痕迹等。

流	组成
KisFlow(1)	KisFunction(V) + KisFunction(S) + KisFunction© + KisFunction(E)
KisFlow(2)	KisFunction(V) + KisFunction(L) + KisFunction(S) + KisFunction© + KisFunction(E)
KisFlow(3)	KisFunction(V) + KisFunction(L) + KisFunction© + KisFunction(E)

通过 KisFunction(S) 和 KisFunction(L)的并流组合关系，各个KisFlow有如下关系：

KisFlow(2) = KisFlow(1) + KisFlow(2)
KisFlow(3) = KisFlow(1) + KisFlow(2) + KisFlow(3)

1.5 KisFlow关键模块

1.5.1 KisConfig

KisConfig为KisFlow的配置模块，其中一个Config包含Flow的配置和Function的配置等。

(1) KisFunction配置

kistype: func
fname: 测试KisFunction_S1
fmode: Save
source:
 name: 被校验的测试数据源1-用户订单维度
 must:
 - userid
 - orderid

option:
 cname: 测试KisConnector_1
 retry_times: 3
 retry_duration: 500
 default_params:
 default1: default1_param
 default2: default2_param

(2) KisFlow配置

kistype: flow
status: 1
flow_name: MyFlow1
flows:
 - fname: 测试PrintInput
 params:
 args1: value1
 args2: value2
 - fname: 测试KisFunction_S1
 - fname: 测试PrintInput
 params:
 args1: value11
 args2: value22
 default2: newDefault
 - fname: 测试PrintInput
 - fname: 测试KisFunction_S1
 params:
 my_user_param1: ffffffxxxxxx
 - fname: 测试PrintInput

KisFlow是一种流式概念形态，具体表现的特征如下：
1、一个KisFlow可以由任意KisFunction组成，且KisFlow可以动态的调整长度。
2、一个KisFunction可以随时动态的加入到某个KisFlow中，且KisFlow和KisFlow之间的关系可以通过KisFunction的Load和Save节点的加入，进行动态的并流和分流动作。
3、KisFlow在编程行为上，从面向流进行数据业务编程，变成了面向KisFunction的函数单计算逻辑的开发，接近FaaS(Function as a service)体系。

(3) KisConnector配置

kistype: conn
cname: 测试KisConnector_1
addrs: '0.0.0.0:9988,0.0.0.0:9999,0.0.0.0:9990'
type: redis
key: userid_orderid_option
params:
 args1: value1
 args2: value2
load: null
save:
 - 测试KisFunction_S1

(4) KisFlow全局配置

#kistype Global为kisflow的全局配置
kistype: global
#是否启动prometheus监控
prometheus_enable: true
#是否需要kisflow单独启动端口监听
prometheus_listen: true
#prometheus取点监听地址
prometheus_serve: 0.0.0.0:20004

1.5.2 KisFunction

KisFunction的类图设计如下

KisFlow快速开始

KisFlow Developer Documentation

Install KisFlow

$go get github.com/aceld/kis-flow

Quick Start

<summary>1. Quick Start </summary>

Source Code

https://github.com/aceld/kis-flow-usage/tree/main/1-quick_start

Project Directory

├── faas_stu_score_avg.go
├── faas_stu_score_avg_print.go
└── main.go

Flow

Main

main.go

package main

import (
	"context"
	"fmt"
	"github.com/aceld/kis-flow/common"
	"github.com/aceld/kis-flow/config"
	"github.com/aceld/kis-flow/flow"
	"github.com/aceld/kis-flow/kis"
)

func main() {
	ctx := context.Background()

	// Create a new flow configuration
	myFlowConfig1 := config.NewFlowConfig("CalStuAvgScore", common.FlowEnable)

	// Create new function configuration
	avgStuScoreConfig := config.NewFuncConfig("AvgStuScore", common.C, nil, nil)
	printStuScoreConfig := config.NewFuncConfig("PrintStuAvgScore", common.E, nil, nil)

	// Create a new flow
	flow1 := flow.NewKisFlow(myFlowConfig1)

	// Link functions to the flow
	_ = flow1.Link(avgStuScoreConfig, nil)
	_ = flow1.Link(printStuScoreConfig, nil)

	// Submit a string
	_ = flow1.CommitRow(`{"stu_id":101, "score_1":100, "score_2":90, "score_3":80}`)
	// Submit a string
	_ = flow1.CommitRow(`{"stu_id":102, "score_1":100, "score_2":70, "score_3":60}`)

	// Run the flow
	if err := flow1.Run(ctx); err != nil {
		fmt.Println("err: ", err)
	}

	return
}

func init() {
	// Register functions
	kis.Pool().FaaS("AvgStuScore", AvgStuScore)
	kis.Pool().FaaS("PrintStuAvgScore", PrintStuAvgScore)
}

Function1

faas_stu_score_avg.go

package main

import (
	"context"
	"github.com/aceld/kis-flow/kis"
	"github.com/aceld/kis-flow/serialize"
)

type AvgStuScoreIn struct {
	serialize.DefaultSerialize
	StuId  int `json:"stu_id"`
	Score1 int `json:"score_1"`
	Score2 int `json:"score_2"`
	Score3 int `json:"score_3"`
}

type AvgStuScoreOut struct {
	serialize.DefaultSerialize
	StuId    int     `json:"stu_id"`
	AvgScore float64 `json:"avg_score"`
}

// AvgStuScore(FaaS) 计算学生平均分
func AvgStuScore(ctx context.Context, flow kis.Flow, rows []*AvgStuScoreIn) error {
	for _, row := range rows {

		out := AvgStuScoreOut{
			StuId:    row.StuId,
			AvgScore: float64(row.Score1+row.Score2+row.Score3) / 3,
		}

		// 提交结果数据
		_ = flow.CommitRow(out)
	}

	return nil
}

Function2

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"`
}

type PrintStuAvgScoreOut struct {
	serialize.DefaultSerialize
}

func PrintStuAvgScore(ctx context.Context, flow kis.Flow, rows []*PrintStuAvgScoreIn) error {

	for _, row := range rows {
		fmt.Printf("stuid: [%+v], avg score: [%+v]\n", row.StuId, row.AvgScore)
	}

	return nil
}

OutPut

Add KisPool FuncName=AvgStuScore
Add KisPool FuncName=PrintStuAvgScore
funcName NewConfig source is nil, funcName = AvgStuScore, use default unNamed Source.
funcName NewConfig source is nil, funcName = PrintStuAvgScore, use default unNamed Source.
stuid: [101], avg score: [90]
stuid: [102], avg score: [76.66666666666667]

Quick Start With Config

<summary>2. Quick Start With Config </summary>

Source Code

https://github.com/aceld/kis-flow-usage/tree/main/2-quick_start_with_config

Project Directory

├── Makefile
├── conf
│   ├── flow-CalStuAvgScore.yml
│   ├── func-AvgStuScore.yml
│   └── func-PrintStuAvgScore.yml
├── faas_stu_score_avg.go
├── faas_stu_score_avg_print.go
└── main.go

Flow

Config

(1) Flow Config

conf/flow-CalStuAvgScore.yml

kistype: flow
status: 1
flow_name: CalStuAvgScore
flows:
  - fname: AvgStuScore
  - fname: PrintStuAvgScore

(2) Function1 Config

conf/func-AvgStuScore.yml

kistype: func
fname: AvgStuScore
fmode: Calculate
source:
  name: StudentScore
  must:
    - stu_id

(3) Function2(Slink) Config

conf/func-PrintStuAvgScore.yml

kistype: func
fname: PrintStuAvgScore
fmode: Expand
source:
  name: StudentScore
  must:
    - stu_id

Main

main.go

package main

import (
	"context"
	"fmt"
	"github.com/aceld/kis-flow/file"
	"github.com/aceld/kis-flow/kis"
)

func main() {
	ctx := context.Background()

	// Load Configuration from file
	if err := file.ConfigImportYaml("conf/"); err != nil {
		panic(err)
	}

	// Get the flow
	flow1 := kis.Pool().GetFlow("CalStuAvgScore")
	if flow1 == nil {
		panic("flow1 is nil")
	}

	// Submit a string
	_ = flow1.CommitRow(`{"stu_id":101, "score_1":100, "score_2":90, "score_3":80}`)
	// Submit a string
	_ = flow1.CommitRow(`{"stu_id":102, "score_1":100, "score_2":70, "score_3":60}`)

	// Run the flow
	if err := flow1.Run(ctx); err != nil {
		fmt.Println("err: ", err)
	}

	return
}

func init() {
	// Register functions
	kis.Pool().FaaS("AvgStuScore", AvgStuScore)
	kis.Pool().FaaS("PrintStuAvgScore", PrintStuAvgScore)
}

Function1

faas_stu_score_avg.go

package main

import (
	"context"
	"github.com/aceld/kis-flow/kis"
	"github.com/aceld/kis-flow/serialize"
)

type AvgStuScoreIn struct {
	serialize.DefaultSerialize
	StuId  int `json:"stu_id"`
	Score1 int `json:"score_1"`
	Score2 int `json:"score_2"`
	Score3 int `json:"score_3"`
}

type AvgStuScoreOut struct {
	serialize.DefaultSerialize
	StuId    int     `json:"stu_id"`
	AvgScore float64 `json:"avg_score"`
}

// AvgStuScore(FaaS) 计
func AvgStuScore(ctx context.Context, flow kis.Flow, rows []*AvgStuScoreIn) error {
	for _, row := range rows {

		out := AvgStuScoreOut{
			StuId:    row.StuId,
			AvgScore: float64(row.Score1+row.Score2+row.Score3) / 3,
		}

		// Commit Result Data
		_ = flow.CommitRow(out)
	}

	return nil
}

Function2

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"`
}

type PrintStuAvgScoreOut struct {
	serialize.DefaultSerialize
}

func PrintStuAvgScore(ctx context.Context, flow kis.Flow, rows []*PrintStuAvgScoreIn) error {

	for _, row := range rows {
		fmt.Printf("stuid: [%+v], avg score: [%+v]\n", row.StuId, row.AvgScore)
	}

	return nil
}

OutPut

Add KisPool FuncName=AvgStuScore
Add KisPool FuncName=PrintStuAvgScore
Add FlowRouter FlowName=CalStuAvgScore
stuid: [101], avg score: [90]
stuid: [102], avg score: [76.66666666666667]

作者:刘丹冰Aceld github: https://github.com/aceld
KisFlow开源项目地址：https://github.com/aceld/kis-flow

点赞
收藏
关注作者

0/1000

抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认取消

加入云驻计划，成为创作者

华为云周边好礼
免费体验产品
特殊身份标识
线下官方门票
内部专家零距离
与10000+优质创作者共同成长

立即加入

Golang框架实战-KisFlow流式计算框架(1)-概述

1.1 为什么需要KisFlow

1.2 KisFlow实要支持的能力

流式计算

分布式任务调度

1.3 KisFlow系统定位

1.4 KisFlow整体架构图

1.5 KisFlow关键模块

1.5.1 KisConfig

(1) KisFunction配置

(2) KisFlow配置

(3) KisConnector配置

(4) KisFlow全局配置

1.5.2 KisFunction

KisFlow快速开始

KisFlow Developer Documentation

Install KisFlow

Source Code

Project Directory

Flow

Main

Function1

Function2

OutPut

Source Code

Project Directory

Flow

Config

(1) Flow Config

(2) Function1 Config

(3) Function2(Slink) Config

Main

Function1

Function2

OutPut

全部回复

设置昵称

关于作者

目录

加入云驻计划，成为创作者

Golang框架实战-KisFlow流式计算框架(1)-概述

1.1 为什么需要KisFlow

1.2 KisFlow实要支持的能力

流式计算

分布式任务调度

1.3 KisFlow系统定位

1.4 KisFlow整体架构图

1.5 KisFlow关键模块

1.5.1 KisConfig

(1) KisFunction配置

(2) KisFlow配置

(3) KisConnector配置

(4) KisFlow全局配置

1.5.2 KisFunction

KisFlow快速开始

KisFlow Developer Documentation

Install KisFlow

Source Code

Project Directory

Flow

Main

Function1

Function2

OutPut

Source Code

Project Directory

Flow

Config

(1) Flow Config

(2) Function1 Config

(3) Function2(Slink) Config

Main

Function1

Function2

OutPut

全部回复

设置昵称

关于作者

目录

加入云驻计划，成为创作者

推荐阅读

相关产品