一种数据流处理方法、系统及计算机设备与流程

1.本发明涉及数据处理技术领域，具体涉及一种数据流处理方法、系统、计算机设备以及非易失性计算机可读存储介质。


背景技术：

2.目前，在日常的爬虫及api开发中，基于整体数据流的处理需求越来越高，有效的流程可以对开发及后续的数据流无论是数据收集、清洗、加工处理及入库起到大幅降低开发成本的作用。现有主流框架例如scrapy往往只能解决数据链中一部分问题，而且功能臃肿，其开发中所遇到的问题例如常见的数据编码、错误通知等功能需要重新开发。因此，设计一种对于数据流从开始收集到入库结束的全流程处理方法来降低开发成本，是本领域技术人员亟待解决的问题。


技术实现要素：

3.鉴于上述现有技术的不足之处，本发明的目的在于提供一种数据流处理方法、系统、计算机设备以及非易失性计算机可读存储介质，可实现业务数据流从开始收集到入库结束的全流程处理，可大幅降低开发成本。
4.为了达到上述目的，本发明采取了以下技术方案：
5.一种数据流处理方法，其中，包括：
6.对业务数据流进行功能任务分配；
7.对各所述功能任务进行参数装载；
8.判断各所述功能任务是否全部完成；
9.对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成；
10.对所述业务数据流进行清洗及处理后插入数据库中。
11.在进一步的技术方案中，所述的数据流处理方法，其中，所述对所述业务数据流进行清洗及处理后插入数据库中之后，包括：
12.对中间数据进行清理以及统一进行资源释放。
13.在进一步的技术方案中，所述的数据流处理方法，其中，所述对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成，包括：
14.对未完成的所述功能任务启动多线程分配子任务执行，根据配置情况为每个线程进行资源发放，直至所述子任务全部完成。
15.在进一步的技术方案中，所述的数据流处理方法，其中，所述对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成，还包括：
16.所述子任务完成后将对应输入输出以键值对的形式存放在redis中。
17.在进一步的技术方案中，所述的数据流处理方法，其中，所述对所述业务数据流进行清洗及处理后插入数据库中，包括：
18.将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中。
19.在进一步的技术方案中，所述的数据流处理方法，其中，所述将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中，所述数据库包括mysql、sqlserver、postgresql。
20.一种数据流处理系统，其中，包括：
21.任务分配模块，用于对业务数据流进行功能任务分配；
22.参数装载模块，用于对各所述功能任务进行参数装载；
23.判断模块，用于判断各所述功能任务是否全部完成；
24.线程分配模块，用于对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成；
25.清洗入库模块，用于对所述业务数据流进行清洗及处理后插入数据库中。
26.在进一步的技术方案中，所述的数据流处理系统，其中，还包括：
27.清理释放模块，用于对中间数据进行清理以及统一进行资源释放。
28.在进一步的技术方案中，所述的数据流处理系统，其中，所述对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成，包括：
29.对未完成的所述功能任务启动多线程分配子任务执行，根据配置情况为每个线程进行资源发放，直至所述子任务全部完成。
30.在进一步的技术方案中，所述的数据流处理系统，其中，所述对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成，还包括：
31.所述子任务完成后将对应输入输出以键值对的形式存放在redis中。
32.在进一步的技术方案中，所述的数据流处理系统，其中，所述对所述业务数据流进行清洗及处理后插入数据库中，包括：
33.将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中。
34.在进一步的技术方案中，所述的数据流处理系统，其中，所述将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中，所述数据库包括mysql、sqlserver、postgresql。
35.一种计算机设备，其中，所述计算机设备包括至少一个处理器；以及，
36.与所述至少一个处理器通信连接的存储器；其中，
37.所述存储器上存储有可被所述至少一个处理器执行的计算机程序，所述计算机程序被所述至少一个处理器执行时，可实现如上述任一项所述的数据流处理方法。
38.一种非易失性计算机可读存储介质，其中，所述非易失性计算机可读存储介质存储有计算机程序，所述计算机程序被至少一个处理器执行时，可实现如上述任一项所述的数据流处理方法。
39.相较于现有技术，本发明提供了一种数据流处理方法、系统及计算机设备，其中，所述方法包括：对业务数据流进行功能任务分配；对各所述功能任务进行参数装载；判断各所述功能任务是否全部完成；对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成；对所述业务数据流进行清洗及处理后插入数据库中。通过本发明的
方法可实现业务数据流从开始收集到入库结束的全流程处理，可大幅降低开发成本。
附图说明
40.为了更清楚地说明本发明实施例或现有技术中的技术方案，下面将对实施例或现有技术描述中所需要使用的附图作简单地介绍，显而易见地，下面描述中的附图仅仅是本发明中记载的一些实施例，对于本领域普通技术人员来讲，在不付出创造性劳动的前提下，还可以根据这些附图获得其它的附图。
41.图1为本发明实施例提供的一种数据流处理方法的流程示意图。
42.图2为本发明另一实施例提供的一种数据流处理方法的流程示意图。
43.图3为本发明实施例提供的一种数据流处理系统的功能模块示意图。
44.图4为本发明另一实施例提供的一种数据流处理系统的功能模块示意图。
45.图5为本发明实施例提供的所述计算机设备的硬件结构示意图。
具体实施方式
46.为使本发明的目的、技术方案及效果更加清楚、明确，以下参照附图并举实施例对本发明进一步详细说明。应当理解，此处所描述的具体实施例仅用以解释本发明，并不用于限定本发明。
47.在本发明的描述中，所使用的“包含”、“包括”、“具有”、“含有”等，均为开放性的用语，即意指包含但不限于。参考术语“一个实施例”、“一个具体实施例”、“一些实施例”、“例如”等的描述意指结合该实施例或示例描述的具体特征、结构或者特点包含于本技术的至少一个实施例或示例中。在本说明书中，对上述术语的示意性表述不一定指的是相同的实施例或示例。而且，描述的具体特征、结构或者特点可以在任何的一个或多个实施例或示例中以合适的方式结合。各实施例中涉及的步骤顺序用于示意性说明本技术的实施，其中的步骤顺序不作限定，可根据需要作适当调整。
48.下面结合附图，详细说明本发明的各种非限制性实施方式。
49.请参阅图1，本发明实施例提供了一种数据流处理方法，其中，包括步骤：
50.s100、对业务数据流进行功能任务分配；
51.s200、对各所述功能任务进行参数装载；
52.s300、判断各所述功能任务是否全部完成；
53.若是，则直接执行步骤s500；
54.若否，则执行步骤s400；
55.s400、对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成；
56.s500、对所述业务数据流进行清洗及处理后插入数据库中。
57.具体地，本实施例中：
58.所述步骤s100、对业务数据流进行功能任务分配；
59.具体实施时，可提前自定义业务数据流的某个节点环境，例如api开发中服务端返回json格式数据流，自动转换为该功能任务所对应的路径及配置条件，初始化设置流程包含的节点例如：数据获取、数据处理(包含业务字段映射等)、入库等。
60.所述步骤s200、对各所述功能任务进行参数装载；
61.具体实施时，预先集成常用工具如requests、httpx、selenium等，基于工厂模式进行的二次开发，参数接口及输出全为统一格式，方便于使用中根据不同场景切换工具而不用改动流程代码。
62.所述步骤s300、判断各所述功能任务是否全部完成；
63.具体实施时，若各所述功能任务全部完成，则直接执行所述步骤s500；若各所述功能任务未全部完成，则执行所述步骤s400。
64.所述步骤s400、对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成；
65.进一步地，所述的数据流处理方法，其中，所述步骤s400、对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成，包括：
66.对未完成的所述功能任务启动多线程分配子任务执行，根据配置情况为每个线程进行资源发放，直至所述子任务全部完成。
67.进一步地，所述的数据流处理方法，其中，所述步骤s400、对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成，还包括：
68.所述子任务完成后将对应输入输出以键值对的形式存放在redis中。
69.具体实施时，对未完成的所述功能任务启动多线程分配子任务执行，根据配置情况为每个线程进行资源发放，即通过当前线程id(唯一)获取对应装载资源，直至所述子任务全部完成；进一步，所述子任务完成后将对应输入输出以键值对的形式存放在redis中，即在数据缓存的过程中，中间数据存放于redis中，输入与输出基于键值对的形式保存，输出以json格式存储，在数据收集的过程中每个键对应1个进度，如出现某个子任务出现错误，则自动将失败的子任务重新放回待完成的分类中，循环调度直到所有子任务全部完成或超过最大循环次数后任务退出，输入输出始终为1对1的关系保障了数据的一致性，任务失败后中间数据仍然会保留最后状态，重启任务后可以无缝衔接之前未完成的状态。
70.所述步骤s500、对所述业务数据流进行清洗及处理后插入数据库中；
71.进一步地，所述的数据流处理方法，其中，所述步骤s500、对所述业务数据流进行清洗及处理后插入数据库中，包括：
72.将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中。
73.进一步地，所述的数据流处理方法，其中，所述将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中，所述数据库包括mysql、sqlserver、postgresql。
74.具体实施时，完成全部任务后，将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中，所述数据库包括mysql、sqlserver、postgresql等，配置数据库连接后，插入的数据不需要列举表头，可以自动将输出数据所有键作为表头存放，个别不同类型列可自定义数据库列类型，默认为字符串类型，插入数据库时会清理已存在同名表保障库中数据与redis中数据的一致性，中间出错则全部数据都不会插入生效，同时提供更新字段设置，即应对不删除表而进行周期更新插入的业务需求。
75.基于上述实施例，请参阅图2，本发明另一实施例还提供了一种数据流处理方法，
其中，所述步骤s500、对所述业务数据流进行清洗及处理后插入数据库中之后，包括步骤：
76.s600、对中间数据进行清理以及统一进行资源释放。
77.具体实施时，即在业务数据流执行完入库操作后，可对中间数据进行清理以及统一进行资源的释放，可减小内存占用。
78.由以上方法实施例可知，本发明提供的数据流处理方法可实现业务数据流从开始收集到入库结束的全流程处理，可大幅降低开发成本。
79.应该理解的是，虽然本技术提供了如实施例或流程图所述的方法操作步骤，但基于常规或者无创造性的劳动可以包括更多或者更少的操作步骤，这些操作步骤并不是必然按照实施例或流程图的顺序依次执行。实施例或流程图中列举的步骤顺序仅仅为众多步骤执行顺序中的一种方式，不代表唯一的执行顺序。而且，实施例或流程图中的至少一部分步骤可以包括多个子步骤或者多个阶段，这些子步骤或者阶段并不必然是在同一时刻执行完成，而是可以在不同的时刻执行，这些子步骤或者阶段的执行顺序也不必然是依次进行，而是可以与其它步骤或者其它步骤的子步骤或者阶段的至少一部分轮流、交替或者同步地执行。
80.基于上述实施例，请参阅图3，本发明另一实施例还提供了一种数据流处理系统，其中，包括：
81.任务分配模块11，用于对业务数据流进行功能任务分配；
82.参数装载模块12，用于对各所述功能任务进行参数装载；
83.判断模块13，用于判断各所述功能任务是否全部完成；
84.线程分配模块14，用于对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成；
85.清洗入库模块15，用于对所述业务数据流进行清洗及处理后插入数据库中。
86.具体地，本实施例中：
87.所述任务分配模块11，用于对业务数据流进行功能任务分配；
88.具体实施时，可提前自定义业务数据流的某个节点环境，例如api开发中服务端返回json格式数据流，自动转换为该功能任务所对应的路径及配置条件，初始化设置流程包含的节点例如：数据获取、数据处理(包含业务字段映射等)、入库等。
89.所述参数装载模块12，用于对各所述功能任务进行参数装载；
90.具体实施时，预先集成常用工具如requests、httpx、selenium等，基于工厂模式进行的二次开发，参数接口及输出全为统一格式，方便于使用中根据不同场景切换工具而不用改动流程代码。
91.所述判断模块13，用于判断各所述功能任务是否全部完成；
92.具体实施时，若各所述功能任务全部完成，则直接执行所述清洗入库模块15；若各所述功能任务未全部完成，则执行所述线程分配模块14。
93.所述线程分配模块14，用于对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成；
94.进一步地，所述的数据流处理系统，其中，所述线程分配模块14，用于对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成，包括：
95.对未完成的所述功能任务启动多线程分配子任务执行，根据配置情况为每个线程
进行资源发放，直至所述子任务全部完成。
96.进一步地，所述的数据流处理系统，其中，所述线程分配模块14，用于对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成，还包括：
97.所述子任务完成后将对应输入输出以键值对的形式存放在redis中。
98.具体实施时，对未完成的所述功能任务启动多线程分配子任务执行，根据配置情况为每个线程进行资源发放，即通过当前线程id(唯一)获取对应装载资源，直至所述子任务全部完成；进一步，所述子任务完成后将对应输入输出以键值对的形式存放在redis中，即在数据缓存的过程中，中间数据存放于redis中，输入与输出基于键值对的形式保存，输出以json格式存储，在数据收集的过程中每个键对应1个进度，如出现某个子任务出现错误，则自动将失败的子任务重新放回待完成的分类中，循环调度直到所有子任务全部完成或超过最大循环次数后任务退出，输入输出始终为1对1的关系保障了数据的一致性，任务失败后中间数据仍然会保留最后状态，重启任务后可以无缝衔接之前未完成的状态。
99.所述清洗入库模块15，用于对所述业务数据流进行清洗及处理后插入数据库中；
100.进一步地，所述的数据流处理系统，其中，所述清洗入库模块15，用于对所述业务数据流进行清洗及处理后插入数据库中，包括：
101.将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中。
102.进一步地，所述的数据流处理系统，其中，所述将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中，所述数据库包括mysql、sqlserver、postgresql。
103.具体实施时，完成全部任务后，将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中，所述数据库包括mysql、sqlserver、postgresql等，配置数据库连接后，插入的数据不需要列举表头，可以自动将输出数据所有键作为表头存放，个别不同类型列可自定义数据库列类型，默认为字符串类型，插入数据库时会清理已存在同名表保障库中数据与redis中数据的一致性，中间出错则全部数据都不会插入生效，同时提供更新字段设置，即应对不删除表而进行周期更新插入的业务需求。
104.基于上述实施例，请参阅图4，本发明另一实施例还提供了一种数据流处理系统，其中，还包括：
105.清理释放模块16，用于对中间数据进行清理以及统一进行资源释放。
106.具体实施时，即在业务数据流执行完入库操作后，可对中间数据进行清理以及统一进行资源的释放，可减小内存占用。
107.由以上系统实施例可知，本发明提供的数据流处理系统可实现业务数据流从开始收集到入库结束的全流程处理，可大幅降低开发成本。
108.基于上述实施例，请参阅图5，本发明另一实施例还提供了一种计算机设备，其中，所述计算机设备10包括：
109.存储器120以及一个或多个处理器110，图5中以一个处理器110为例进行介绍，处理器110和存储器120可以通过通信总线或者其他方式连接，图5中以通过通信总线连接为例。
110.处理器110用于完成计算机设备10的各种控制逻辑，其可以为通用处理器、数字信
号处理器(dsp)、专用集成电路(asic)、现场可编程门阵列(fpga)、单片机、arm(acorn risc machine)或其它可编程逻辑器件、分立门或晶体管逻辑、分立的硬件组件或者这些部件的任何组合。还有，处理器110还可以是任何传统处理器、微处理器或状态机。处理器110也可以被实现为计算设备的组合，例如，dsp和微处理器的组合、多个微处理器、一个或多个微处理器结合dsp核、或任何其它这种配置。
111.存储器120作为一种非易失性计算机可读存储介质，可用于存储非易失性软件程序、非易失性计算机可执行程序以及模块，如本发明实施例中的数据流处理方法对应的计算机程序。处理器110通过运行存储在存储器120中的非易失性软件程序、指令以及单元，从而执行计算机设备10的各种功能应用以及数据处理，即实现上述方法实施例中的数据流处理方法。
112.存储器120可以包括存储程序区和存储数据区，其中，存储程序区可存储操作装置、至少一个功能所需要的应用程序；存储数据区可存储根据计算机设备10使用所创建的数据等。此外，存储器120可以包括高速随机存取存储器，还可以包括非易失性存储器，例如至少一个磁盘存储器件、闪存器件、或其他非易失性固态存储器件。在一些实施例中，存储器120可选包括相对于处理器110远程设置的存储器，这些远程存储器可以通过网络连接至计算机设备10。上述网络的实例包括但不限于互联网、企业内部网、局域网、移动通信网及其组合。
113.一个或者多个单元存储在存储器120中，当被一个或者多个处理器110执行时，可实现如上述任一项方法实施例中的数据流处理方法，例如，可实现以上描述的图1中的方法步骤s100至步骤s500。
114.本领域技术人员可以理解，图5中示出的硬件结构示意图，仅仅是与本发明方案相关的部分结构的示意图，并不构成对本发明方案所应用于其上的计算机设备的限定，具体的计算机设备可以包括比图中所示更多的部件，或者组合某些部件，或者具有不同的部件布置。
115.基于上述实施例，本发明还提供了一种非易失性计算机可读存储介质，其中，所述非易失性计算机可读存储介质存储有计算机程序，所述计算机程序被至少一个处理器执行时，可实现如上述任一项方法实施例中的数据流处理方法，例如，可实现以上描述的图1中的方法步骤s100至步骤s500。
116.作为示例，非易失性存储介质能够包括只读存储器(rom)、可编程rom(prom)、电可编程rom(eprom)、电可擦rom(eeprom)或闪速存储器。易失性存储器能够包括作为外部高速缓存存储器的随机存取存储器(ram)。通过说明并非限制，ram可以以诸如同步ram(sram)、动态ram、(dram)、同步dram(sdram)、双数据速率sdram(ddr sdram)、增强型sdram(esdram)、synchl ink dram(sldram)以及直接rambus(兰巴斯)ram(drram)之类的许多形式得到。本文中所描述的操作环境的所公开的存储器组件或存储器旨在包括这些和/或任何其他适合类型的存储器中的一个或多个。
117.本发明的另一种实施例提供了一种计算机程序产品，所述计算机程序产品包括存储在非易失性计算机可读存储介质上的计算机程序，所述计算机程序包括程序指令，当所述程序指令被处理器执行时，可实现如上述任一项方法实施例中的数据流处理方法，例如，可实现以上描述的图1中的方法步骤s100至步骤s500。
118.以上所描述的实施例仅仅是示意性的，其中作为分离部件说明的单元可以是或者也可以不是物理上分开的，作为单元显示的部件可以是或者也可以不是物理单元，即可以位于一个地方，或者也可以分布到多个网络单元上。可以根据实际需要选择其中的部分或者全部模块来实现本实施例方案的目的。
119.通过以上的实施例的描述，本领域的技术人员可以清楚地了解到各实施例可借助软件加通用硬件平台的方式来实现，当然也可以通过硬件实现。基于这样的理解，上述技术方案本质上或者说对相关技术做出贡献的部分可以以软件产品的形式体现出来，该计算机软件产品可以存在于计算机可读存储介质中，如rom/ram、磁碟、光盘等，包括若干指令用以使得一台计算机装置(可以是个人计算机，服务器，或者网络装置等)执行各个实施例或者实施例的某些部分的方法。
120.除了其他之外，诸如“能够”、“能”、“可能”或“可以”之类的条件语言除非另外具体地陈述或者在如所使用的上下文内以其他方式理解，否则一般地旨在传达特定实施方式能包括(然而其他实施方式不包括)特定特征、元件和/或操作。因此，这样的条件语言一般地还旨在暗示特征、元件和/或操作对于一个或多个实施方式无论如何都是需要的或者一个或多个实施方式必须包括用于在有或没有输入或提示的情况下判定这些特征、元件和/或操作是否被包括或者将在任何特定实施方式中被执行的逻辑。
121.已经在本文中在本说明书和附图中描述的内容包括能够提供数据流处理方法、系统、计算机设备以及非易失性计算机可读存储介质的示例。当然，不能够出于描述本公开的各种特征的目的来描述元件和/或方法的每个可以想象的组合，但是可以认识到，所公开的特征的许多另外的组合和置换是可能的。因此，显而易见的是，在不脱离本公开的范围或精神的情况下能够对本公开做出各种修改，但是，所有这些各种修改都应属于本发明所附权利要求的保护范围。此外，或在替代方案中，本公开的其他实施例从对本说明书和附图的考虑以及如本文中所呈现的本公开的实践中可能是显而易见的。意图是，本说明书和附图中所提出的示例在所有方面被认为是说明性的而非限制性的。尽管在本文中采用了特定术语，但是它们在通用和描述性意义上被使用并且不用于限制的目的。

技术特征：
1.一种数据流处理方法，其特征在于，包括：对业务数据流进行功能任务分配；对各所述功能任务进行参数装载；判断各所述功能任务是否全部完成；对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成；对所述业务数据流进行清洗及处理后插入数据库中。2.根据权利要求1所述的数据流处理方法，其特征在于，所述对所述业务数据流进行清洗及处理后插入数据库中之后，包括：对中间数据进行清理以及统一进行资源释放。3.根据权利要求2所述的数据流处理方法，其特征在于，所述对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成，包括：对未完成的所述功能任务启动多线程分配子任务执行，根据配置情况为每个线程进行资源发放，直至所述子任务全部完成。4.根据权利要求3所述的数据流处理方法，其特征在于，所述对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成，还包括：所述子任务完成后将对应输入输出以键值对的形式存放在redis中。5.根据权利要求4所述的数据流处理方法，其特征在于，所述对所述业务数据流进行清洗及处理后插入数据库中，包括：将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中。6.根据权利要求5所述的数据流处理方法，其特征在于，所述将所述业务数据流中的数据取出，并进行清洗加工后提取所有字段根据配置情况插入数据库中，所述数据库包括mysql、sqlserver、postgresql。7.一种数据流处理系统，其特征在于，包括：任务分配模块，用于对业务数据流进行功能任务分配；参数装载模块，用于对各所述功能任务进行参数装载；判断模块，用于判断各所述功能任务是否全部完成；线程分配模块，用于对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成；清洗入库模块，用于对所述业务数据流进行清洗及处理后插入数据库中。8.根据权利要求7所述的数据流处理系统，其特征在于，还包括：清理释放模块，用于对中间数据进行清理以及统一进行资源释放。9.一种计算机设备，其特征在于，所述计算机设备包括至少一个处理器；以及，与所述至少一个处理器通信连接的存储器；其中，所述存储器上存储有可被所述至少一个处理器执行的计算机程序，所述计算机程序被所述至少一个处理器执行时，可实现如权利要求1-6任一项所述的数据流处理方法。10.一种非易失性计算机可读存储介质，其特征在于，所述非易失性计算机可读存储介质存储有计算机程序，所述计算机程序被至少一个处理器执行时，可实现如权利要求1-6任一项所述的数据流处理方法。

技术总结
本发明公开了一种数据流处理方法、系统及计算机设备，涉及数据处理技术领域，其中，所述方法包括：对业务数据流进行功能任务分配；对各所述功能任务进行参数装载；判断各所述功能任务是否全部完成；对未完成的所述功能任务启动多线程分配子任务执行，直至所述子任务全部完成；对所述业务数据流进行清洗及处理后插入数据库中。通过本发明的方法可实现业务数据流从开始收集到入库结束的全流程处理，可大幅降低开发成本。低开发成本。低开发成本。


技术研发人员：赵少玲 赵松耿 刘建豪
受保护的技术使用者：深圳市瞳学科技有限公司
技术研发日：2022.10.27
技术公布日：2023/1/6