一、概述

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 整体架构图#

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

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

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_paramCopy

(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: 测试PrintInputCopy

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_S1Copy

(4) KisFlow 全局配置#

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

1.5.2 KisFunction#

KisFunction 的类图设计如下

作者：刘丹冰 Aceld github: github.com/aceld

KisFlow 开源项目地址：github.com/aceld/kis-flow