一种基于编排的数据处理方法和装置与流程

本技术涉及数据处理，尤其涉及一种基于编排的数据处理方法和装置。

背景技术：

1、随着数字化转型的深入，企业在数据处理领域面临着日益复杂的混合业务需求。在许多实际业务场景中，系统不仅需要具备对实时数据流（如用户行为、设备状态）进行低延迟响应的能力，以满足即时监控与预警的需求，同时也需要能够高效地执行周期性的批量处理任务（如日报表统计、历史数据归档），以支撑深度分析与长期规划。

2、目前，针对上述混合业务需求，业界普遍采用的是流式处理与批式处理独立部署的架构方案。具体而言，对于处理实时数据流的实时任务，通常通过kafka等消息队列接入，并由apache flink等流式计算引擎完成低延迟的流式处理；而对于处理批量数据的调度任务，则依赖apache airflow等定时调度工具，在预设的时间点触发离线计算作业。在这种架构下，两类任务的流程逻辑彼此割裂，各自拥有独立的数据处理组件与代码实现，互不相通。因此，实时任务与调度任务的数据处理组件无法共享复用，导致每当业务规则发生变更时，开发人员需要分别在流式作业和批式脚本中同步修改代码逻辑，不仅大幅增加了开发与维护成本，还极易因修改不同步而引发数据口径不一致的问题。这使得系统难以满足用户快速实现业务中实时监控和周期性分析的混合需求。

3、因此，亟需一种能够满足用户快速实现业务中实时监控和周期性分析的混合需求的方案。

技术实现思路

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、进一步地，基于本技术的前端可视化编排界面，当业务规则发生变化时，用户只需通过可视化编排界面修改配置文件并重新提交，无需分别修改流式和批式的代码逻辑，更无需分别编译部署两套系统，大幅降低了开发与维护成本。后台可基于最新的配置文件重新创建相关任务实例，并停止旧的任务实例，实现了任务实例的动态更新，保证了对业务逻辑变更的快速响应。