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

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
可观测可视化 Grafana 版,10个用户账号 1个月
简介: KisFlow是针对缺乏数仓平台但又有实时计算需求的企业的解决方案,它提供分布式批量消费、有状态流式计算、数据流监控和分布式任务调度等功能。通过KisFunction实现业务逻辑复用,减轻对业务数据库的压力。系统包括流式计算层和任务调度层,支持多种数据源和中间件集成。KisConfig用于配置管理,KisFunction是基本计算单元。设计目标是使业务工程师能轻松进行流式计算。项目源码可在GitHub查看:https://github.com/aceld/kis-flow。

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


1.1 为什么需要KisFlow

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

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

image.png

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

1.2 KisFlow实要支持的能力

流式计算

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

分布式任务调度

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

1.3 KisFlow系统定位

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

image.png

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: 根据业务划分的逻辑任务集合。
调度日志及用户管理:任务调度日志收集、调度详细、调度流程痕迹等。

image.png

组成
KisFlow(1) KisFunction(V) + KisFunction(S) + KisFunction(C) + KisFunction(E)
KisFlow(2) KisFunction(V) + KisFunction(L) + KisFunction(S) + KisFunction(C) + KisFunction(E)
KisFlow(3) KisFunction(V) + KisFunction(L) + KisFunction(C) + 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

image.png

image.png

(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

image.png

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

image.png

(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

image.png

(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的类图设计如下

image.png


作者:刘丹冰Aceld github: https://github.com/aceld
技术主页: https://yuque.com/aceld

相关实践学习
数据库实验室挑战任务-初级任务
本场景介绍如何开通属于你的免费云数据库,在RDS-MySQL中完成对学生成绩的详情查询,执行指定类型SQL。
阿里云云原生数据仓库AnalyticDB MySQL版 使用教程
云原生数据仓库AnalyticDB MySQL版是一种支持高并发低延时查询的新一代云原生数据仓库,高度兼容MySQL协议以及SQL:92、SQL:99、SQL:2003标准,可以对海量数据进行即时的多维分析透视和业务探索,快速构建企业云上数据仓库。 了解产品 https://www.aliyun.com/product/ApsaraDB/ads
目录
相关文章
|
13天前
|
SQL 关系型数据库 MySQL
探索Gorm - Golang流行的数据库ORM框架
探索Gorm - Golang流行的数据库ORM框架
|
11天前
|
NoSQL 测试技术 Go
【Golang】国密SM2公钥私钥序列化到redis中并加密解密实战_sm2反编(1)
【Golang】国密SM2公钥私钥序列化到redis中并加密解密实战_sm2反编(1)
|
13天前
|
Go
golang学习3,golang 项目中配置gin的web框架
golang学习3,golang 项目中配置gin的web框架
|
13天前
|
分布式计算 Java Go
Golang深入浅出之-Go语言中的分布式计算框架Apache Beam
【5月更文挑战第6天】Apache Beam是一个统一的编程模型,适用于批处理和流处理,主要支持Java和Python,但也提供实验性的Go SDK。Go SDK的基本概念包括`PTransform`、`PCollection`和`Pipeline`。在使用中,需注意类型转换、窗口和触发器配置、资源管理和错误处理。尽管Go SDK文档有限,生态系统尚不成熟,且性能可能不高,但它仍为分布式计算提供了可移植的解决方案。通过理解和掌握Beam模型,开发者能编写高效的数据处理程序。
143 1
|
13天前
|
JSON 监控 安全
Golang深入浅出之-Go语言中的反射(reflect):原理与实战应用
【5月更文挑战第1天】Go语言的反射允许运行时检查和修改结构,主要通过`reflect`包的`Type`和`Value`实现。然而,滥用反射可能导致代码复杂和性能下降。要安全使用,应注意避免过度使用,始终进行类型检查,并尊重封装。反射的应用包括动态接口实现、JSON序列化和元编程。理解反射原理并谨慎使用是关键,应尽量保持代码静态类型。
28 2
|
13天前
|
JSON JavaScript 前端开发
Golang深入浅出之-Go语言JSON处理:编码与解码实战
【4月更文挑战第26天】本文探讨了Go语言中处理JSON的常见问题及解决策略。通过`json.Marshal`和`json.Unmarshal`进行编码和解码,同时指出结构体标签、时间处理、omitempty使用及数组/切片区别等易错点。建议正确使用结构体标签,自定义处理`time.Time`,明智选择omitempty,并理解数组与切片差异。文中提供基础示例及时间类型处理的实战代码,帮助读者掌握JSON操作。
24 1
Golang深入浅出之-Go语言JSON处理:编码与解码实战
|
13天前
|
XML JSON 人工智能
探索Gin框架:Golang Gin框架请求参数的获取
探索Gin框架:Golang Gin框架请求参数的获取
|
13天前
|
存储 人工智能 Go
探索Gin框架:Golang使用Gin完成文件上传
探索Gin框架:Golang使用Gin完成文件上传
|
13天前
|
存储 中间件 Go
探索Gin框架:快速构建高性能的Golang Web应用
探索Gin框架:快速构建高性能的Golang Web应用
|
13天前
|
存储 测试技术 Go
Golang框架实战-KisFlow流式计算框架(2)-项目构建/基础模块-(上)
KisFlow项目源码位于<https://github.com/aceld/kis-flow,初始阶段涉及项目构建和基础模块定义。首先在GitHub创建仓库,克隆到本地。项目目录包括`common/`, `example/`, `function/`, `conn/`, `config/`, `flow/`, 和 `kis/`。`go.mod`用于包管理,`KisLogger`接口定义了日志功能,提供不同级别的日志方法。默认日志对象`kisDefaultLogger`打印到标准输出。
625 1
Golang框架实战-KisFlow流式计算框架(2)-项目构建/基础模块-(上)