一种无侵入式的数据抽取加工方法及系统与流程

1.本发明涉及数据处理技术领域，尤其涉及一种无侵入式的数据抽取加工方法及系统。


背景技术：

2.目前，在信息化技术中，大多使用微服务理念将复杂的业务系统通过业务边界拆分为多个系统，而每个系统有各自的数据库，各系统通过网络通信进行交互数据最终完成业务的全流程逻辑处理。
3.在上述数据架构中，通常采用传统的联表方式进行数据加工，但联表方式已经无法适用这种数据架构，特别是在海量数据情况下无法实现实时数据计算处理。
4.因此，需要提出一种新的数据处理方法，以应对多业务系统和高实时性的数据处理需求。


技术实现要素：

5.本发明提供一种无侵入式的数据抽取加工方法及系统，用以解决现有技术中针对多业务系统的海量实时数据处理需求，无法实现高实时性和无感知的缺陷。
6.第一方面，本发明提供一种无侵入式的数据抽取加工方法，包括：
7.利用增量订阅消费组件提取多个业务系统的增量数据；
8.通过jdbc获取多个业务系统的存量历史数据；
9.对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据；
10.对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用。
11.根据本发明提供的一种无侵入式的数据抽取加工方法，利用增量订阅消费组件提取多个业务系统的增量数据，包括：
12.采用canal同步监听获取多个业务系统中数据库的binlog数据；
13.将binlog数据发送至消息中间件kafka。
14.根据本发明提供的一种无侵入式的数据抽取加工方法，通过jdbc获取多个业务系统的存量历史数据，包括：
15.基于预设配置方式配置业务系统数据库，确定数据库中待同步表名的dml和表对应主键id；
16.根据所述dml和所述表对应主键id，通过jdbc获取排序后的数据库表中的存量历史数据；
17.将所述存量历史数据发送至消息中间件kafka。
18.根据本发明提供的一种无侵入式的数据抽取加工方法，基于预设配置方式配置业务系统数据库，包括：
19.通过对所述业务系统数据库的页面进行增加、删除、查找和修改，配置待同步的数
据库账号、密码、表格和主键字段。
20.根据本发明提供的一种无侵入式的数据抽取加工方法，对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据，包括：
21.通过kafka消息数据将所述增量数据和所述存量历史数据存储至elasticsearch中，根据业务系统数据库中表名创建索引表，得到所述加工后数据；
22.其中，所述索引表用于进行数据验证和查询。
23.根据本发明提供的一种无侵入式的数据抽取加工方法，对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据，还包括：
24.通过kafka消息数据将所述增量数据和所述存量历史数据按照预设业务维度进行数据存储；
25.以所述预设业务维度中任一维度进行数据聚合，将其他数据按照json格式存入elasticsearch中，输出任一维度预设大宽表；
26.将多个业务系统中任一维度的数据根据业务逻辑聚合，并按照数组形式存储至所述任一维度预设大宽表的任一业务字段中，得到所述加工后数据。
27.根据本发明提供的一种无侵入式的数据抽取加工方法，对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用，包括：
28.确定存储数据库为elasticsearch；
29.获取多个业务系统的加工后数据中每个字段数据的聚合计算结果，利用去重标识调用存储原始数据。
30.根据本发明提供的一种无侵入式的数据抽取加工方法，利用去重标识调用存储原始数据，包括：
31.将已调用消费数据的主键id存入redis，确定预设有效期；
32.在所述预设有效期内调用数据时将所述redis中的主键id与当前主键id进行对比，若确定不一致，则继续调用，否则停止调用。
33.第二方面，本发明还提供一种无侵入式的数据抽取加工系统，包括：
34.增量提取模块，用于利用增量订阅消费组件提取多个业务系统的增量数据；
35.存量提取模块，用于通过jdbc获取多个业务系统的存量历史数据；
36.加工处理模块，用于对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据；
37.数据调用模块，用于对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用。
38.第三方面，本发明还提供一种电子设备，包括存储器、处理器及存储在存储器上并可在处理器上运行的计算机程序，所述处理器执行所述程序时实现如上述任一种所述无侵入式的数据抽取加工方法。
39.第四方面，本发明还提供一种非暂态计算机可读存储介质，其上存储有计算机程序，该计算机程序被处理器执行时实现如上述任一种所述无侵入式的数据抽取加工方法。
40.本发明提供的无侵入式的数据抽取加工方法及系统，通过对多种业务系统采用无侵入感知获取业务系统增量及全量数据和实时加工处理数据，实现了实时计算的数据存
储、历史数据回溯处理以及支持海量数据快速查询，无需对业务系统进行改造即可进行数据综合管理，具有高实时性和高效率。
附图说明
41.为了更清楚地说明本发明或现有技术中的技术方案，下面将对实施例或现有技术描述中所需要使用的附图作简单地介绍，显而易见地，下面描述中的附图是本发明的一些实施例，对于本领域普通技术人员来讲，在不付出创造性劳动的前提下，还可以根据这些附图获得其他的附图。
42.图1是本发明提供的无侵入式的数据抽取加工方法的流程示意图之一；
43.图2是本发明提供的无侵入式的数据抽取加工方法的流程示意图之二；
44.图3是本发明提供的实时计算数据存储示意图；
45.图4是本发明提供的无侵入式的数据抽取加工系统的结构示意图；
46.图5是本发明提供的电子设备的结构示意图。
具体实施方式
47.为使本发明的目的、技术方案和优点更加清楚，下面将结合本发明中的附图，对本发明中的技术方案进行清楚、完整地描述，显然，所描述的实施例是本发明一部分实施例，而不是全部的实施例。基于本发明中的实施例，本领域普通技术人员在没有做出创造性劳动前提下所获得的所有其他实施例，都属于本发明保护的范围。
48.图1是本发明实施例提供的无侵入式的数据抽取加工方法的流程示意图之一，如图1所示，包括：
49.步骤100：利用增量订阅消费组件提取多个业务系统的增量数据；
50.步骤200：通过jdbc获取多个业务系统的存量历史数据；
51.步骤300：对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据；
52.步骤400：对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用。
53.本发明实施例针对多个业务系统的数据处理，首先通过无侵入业务系统提取业务系统增量数据，并提取业务系统的存量历史数据，然后针对提取的增量数据和存量历史数据进行实时加工处理数据，最后对加工后的数据进行存储，并提供给业务系统进行实时调用。
54.具体地，如图2所示，各业务系统的数据库是独立设置的，如数据库a、数据库b、数据库c、数据库d
……
，一般为mysql数据库，一方面利用增量订阅消费组件canal提取上述多个数据库中的日志数据，如binlog数据，将binlog数据发送至消息中间件kafka，完成增量数据的数据订阅，另一方面通过java数据库连接(java database connectivity，jdbc)遍历各数据库的表数据，进行存量历史数据提取，也是通过消息中间件kafka进行发送。待完成数据加工处理后，将加工后数据搜索与数据分析引擎elasticsearch中，实现数据入库，通过存储数据库可查询实时数据。
55.本发明通过对多种业务系统采用无侵入感知获取业务系统增量及全量数据和实
时加工处理数据，实现了实时计算的数据存储、历史数据回溯处理以及支持海量数据快速查询，无需对业务系统进行改造即可进行数据综合管理，具有高实时性和高效率。
56.在上述实施例的基础上，利用增量订阅消费组件提取多个业务系统的增量数据，包括：
57.采用canal同步监听获取多个业务系统中数据库的binlog数据；
58.将binlog数据发送至消息中间件kafka。
59.具体地，本发明实施例通过增量订阅消费组件canal同步各业务系统数据库的binlog数据，将binlog数据发送到消息中间件kafka。
60.此处，canal是阿里巴巴mysql数据库binlog的增量订阅消费组件，可支持的业务包括：数据库镜像、数据库实时备份、多级索引、业务cache刷新以及价格变化等重要业务消息。
61.binlog数据即二进制日志binary log，包含描述数据库更改的“事件”，例如表创建操作或对表数据的更改。mysql有两层结构，第一层是server层，里面包含连接器、查询缓存、解析器、优化器和执行器，第二层是存储引擎层，例如：innodb、myisam、memory等多个存储引擎。binlog数据产生于mysql中的server层。
62.消息中间件kafka是由apache软件基金会开发的一个开源流处理平台，由scala和java编写。kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。
63.在上述实施例的基础上，通过jdbc获取多个业务系统的存量历史数据，包括：
64.基于预设配置方式配置业务系统数据库，确定数据库中待同步表名的数据操纵语句dml和表对应主键id；
65.根据所述dml和所述表对应主键id，通过jdbc获取排序后的数据库表中的存量历史数据；
66.将所述存量历史数据发送至消息中间件kafka。
67.其中，基于预设配置方式配置业务系统数据库，包括：
68.通过对所述业务系统数据库的页面进行增加、删除、查找和修改，配置待同步的数据库账号、密码、表格和主键字段。
69.具体地，本发明实施例通过预先设置好的配置方式配置需要同步业务系统数据库的账号密码，这里为只读权限，定义好需要同步表名的数据操纵语句(data manipulation language，dml)和表对应的主键id，通过jdbc操作表的分页，这里是通过id进行排序，读取历史存量数据，发送至消息中间件kafka。
70.上述配置方式包括数据库表配置需要同步的数据库账号、密码、表和主键字段等，通过管理页面支持配置数据的增加、删除、查找和修改等操作。
71.dml用于数据库操作，对数据库其中的对象和数据运行访问工作的编程语句，通常是数据库专用编程语言之中的一个子集，以insert、update、delete三种指令为核心，分别代表插入、更新与删除，是开发以数据为中心的应用程序必定会使用到的指令，另外还加上select指令。
72.在上述实施例的基础上，对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据，包括：
73.通过kafka消息数据将所述增量数据和所述存量历史数据存储至elasticsearch中，根据业务系统数据库中表名创建索引表，得到所述加工后数据；
74.其中，所述索引表用于进行数据验证和查询。
75.其中，对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据，还包括：
76.通过kafka消息数据将所述增量数据和所述存量历史数据按照预设业务维度进行数据存储；
77.以所述预设业务维度中任一维度进行数据聚合，将其他数据按照json格式存入elasticsearch中，输出任一维度预设大宽表；
78.将多个业务系统中任一维度的数据根据业务逻辑聚合，并按照数组形式存储至所述任一维度预设大宽表的任一业务字段中，得到所述加工后数据。
79.具体地，本发明实施例针对获取的增量数据和存量历史数据进行实时加工处理，该实时数据处理加工分为两种模式：
80.第一种模式是消费kafka消息数据将原始数据存入搜索与数据分析引擎elasticsearch中，通过数据库中表名创建索引index。原始表支持业务进行数据验证和一般查询操作。
81.elasticsearch是一个分布式、高扩展以及高实时的搜索与数据分析引擎。它能很方便的使大量数据具有搜索、分析和探索的能力，充分利用elasticsearch的水平伸缩性，能使数据在生产环境变得更有价值，elasticsearch的实现原理主要分为以下几个步骤，首先用户将数据提交到elasticsearch数据库中，再通过分词控制器去将对应的语句分词，将其权重和分词结果一并存入数据，当用户搜索数据时候，再根据权重将结果排名，打分，再将返回结果呈现给用户。
82.第二种模式是消费kafka消息数据以业务维度聚合数据存储，比如将以客户号为维度聚合一条数据，数据上的其他字段通过json格式存入elasticsearch，形成客户号为维度或某种业务维度的大宽表，例如图3所示的实时计算数据存储方式示意图，不同的业务字段a、b、c通过json格式向elasticsearch存储数据，消费kafka消息时将多个业务系统中客户号或某种业务维度(能关联各数据的一个编号，例如客户号、身份证号、统一流水号和订单号等能关联多个业务表数据的字段)的数据按照业务逻辑聚合以数组的形式存储至宽表的某个业务字段中。
83.在上述实施例的基础上，对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用，包括：
84.确定存储数据库为elasticsearch；
85.获取多个业务系统的加工后数据中每个字段数据的聚合计算结果，利用去重标识调用存储原始数据。
86.其中，利用去重标识调用存储原始数据，包括：
87.将已调用消费数据的主键id存入远程字典服务redis，确定预设有效期；
88.在所述预设有效期内调用数据时将所述redis中的主键id与当前主键id进行对比，若确定不一致，则继续调用，否则停止调用。
89.具体地，本发明实施例针对加工后的数据进行存储并提供给业务系统进行实时调
用时，存储选用elasticsearch作为存储数据库。
90.在数据存储时，每个字段中的数据若是聚合计算结果则应存储聚合计算结果并保存原始数据，这里的原始数据需要通过去重标识进行区分，避免重复消费导致数据不准确。
91.需要说明的是，这里的去重标识是通过已经消费过的数据将主键id存入redis，此处还需要设置一个有效期，每次消费时去redis查询下是否已经消费过，这样可以避免短时间内的重复消费。对于数据多次回溯导致的重复消费，根据数据业务状态来判断本条数据是否已经消费过，避免了重复消费调用数据。
92.例如，业务系统如果需要查询某个客户的某种行为次数，只需要通过客户号到客户维度宽表中查询行为字段的数据，拿到数据后进行简单的聚合计算即可得到查询结果。
93.下面对本发明提供的无侵入式的数据抽取加工系统进行描述，下文描述的无侵入式的数据抽取加工系统与上文描述的无侵入式的数据抽取加工方法可相互对应参照。
94.图4是本发明实施例提供的无侵入式的数据抽取加工系统的结构示意图，如图4所示，包括：增量提取模块41、存量提取模块42、加工处理模块43和数据调用模块44，其中：
95.增量提取模块41用于利用增量订阅消费组件提取多个业务系统的增量数据；存量提取模块42用于通过jdbc获取多个业务系统的存量历史数据；加工处理模块43用于对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据；数据调用模块44用于对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用。
96.图5示例了一种电子设备的实体结构示意图，如图5所示，该电子设备可以包括：处理器(processor)510、通信接口(communications interface)520、存储器(memory)530和通信总线540，其中，处理器510，通信接口520，存储器530通过通信总线540完成相互间的通信。处理器510可以调用存储器530中的逻辑指令，以执行无侵入式的数据抽取加工方法，该方法包括：利用增量订阅消费组件提取多个业务系统的增量数据；通过jdbc获取多个业务系统的存量历史数据；对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据；对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用。
97.此外，上述的存储器530中的逻辑指令可以通过软件功能单元的形式实现并作为独立的产品销售或使用时，可以存储在一个计算机可读取存储介质中。基于这样的理解，本发明的技术方案本质上或者说对现有技术做出贡献的部分或者该技术方案的部分可以以软件产品的形式体现出来，该计算机软件产品存储在一个存储介质中，包括若干指令用以使得一台计算机设备(可以是个人计算机，服务器，或者网络设备等)执行本发明各个实施例所述方法的全部或部分步骤。而前述的存储介质包括：u盘、移动硬盘、只读存储器(rom，read-only memory)、随机存取存储器(ram，random access memory)、磁碟或者光盘等各种可以存储程序代码的介质。
98.另一方面，本发明还提供一种计算机程序产品，所述计算机程序产品包括计算机程序，计算机程序可存储在非暂态计算机可读存储介质上，所述计算机程序被处理器执行时，计算机能够执行上述各方法所提供的无侵入式的数据抽取加工方法，该方法包括：利用增量订阅消费组件提取多个业务系统的增量数据；通过jdbc获取多个业务系统的存量历史数据；对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据；对所述加
工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用。
99.又一方面，本发明还提供一种非暂态计算机可读存储介质，其上存储有计算机程序，该计算机程序被处理器执行时实现以执行上述各方法提供的无侵入式的数据抽取加工方法，该方法包括：利用增量订阅消费组件提取多个业务系统的增量数据；通过jdbc获取多个业务系统的存量历史数据；对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据；对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用。
100.以上所描述的装置实施例仅仅是示意性的，其中所述作为分离部件说明的单元可以是或者也可以不是物理上分开的，作为单元显示的部件可以是或者也可以不是物理单元，即可以位于一个地方，或者也可以分布到多个网络单元上。可以根据实际的需要选择其中的部分或者全部模块来实现本实施例方案的目的。本领域普通技术人员在不付出创造性的劳动的情况下，即可以理解并实施。
101.通过以上的实施方式的描述，本领域的技术人员可以清楚地了解到各实施方式可借助软件加必需的通用硬件平台的方式来实现，当然也可以通过硬件。基于这样的理解，上述技术方案本质上或者说对现有技术做出贡献的部分可以以软件产品的形式体现出来，该计算机软件产品可以存储在计算机可读存储介质中，如rom/ram、磁碟、光盘等，包括若干指令用以使得一台计算机设备(可以是个人计算机，服务器，或者网络设备等)执行各个实施例或者实施例的某些部分所述的方法。
102.最后应说明的是：以上实施例仅用以说明本发明的技术方案，而非对其限制；尽管参照前述实施例对本发明进行了详细的说明，本领域的普通技术人员应当理解：其依然可以对前述各实施例所记载的技术方案进行修改，或者对其中部分技术特征进行等同替换；而这些修改或者替换，并不使相应技术方案的本质脱离本发明各实施例技术方案的精神和范围。

技术特征：
1.一种无侵入式的数据抽取加工方法，其特征在于，包括：利用增量订阅消费组件提取多个业务系统的增量数据；通过java数据库连接jdbc获取多个业务系统的存量历史数据；对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据；对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用。2.根据权利要求1所述的无侵入式的数据抽取加工方法，其特征在于，利用增量订阅消费组件提取多个业务系统的增量数据，包括：采用增量订阅消费组件canal同步监听获取多个业务系统中数据库的二进制日志binlog数据；将binlog数据发送至消息中间件分布式发布订阅消息系统kafka。3.根据权利要求1所述的无侵入式的数据抽取加工方法，其特征在于，通过jdbc获取多个业务系统的存量历史数据，包括：基于预设配置方式配置业务系统数据库，确定数据库中待同步表名的数据操纵语句dml和表对应主键id；根据所述dml和所述表对应主键id，通过jdbc获取排序后的数据库表中的存量历史数据；将所述存量历史数据发送至消息中间件kafka。4.根据权利要求3所述的无侵入式的数据抽取加工方法，其特征在于，基于预设配置方式配置业务系统数据库，包括：通过对所述业务系统数据库的页面进行增加、删除、查找和修改，配置待同步的数据库账号、密码、表格和主键字段。5.根据权利要求1所述的无侵入式的数据抽取加工方法，其特征在于，对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据，包括：通过kafka消息数据将所述增量数据和所述存量历史数据存储至搜索与数据分析引擎elasticsearch中，根据业务系统数据库中表名创建索引表，得到所述加工后数据；其中，所述索引表用于进行数据验证和查询。6.根据权利要求5所述的无侵入式的数据抽取加工方法，其特征在于，对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据，还包括：通过kafka消息数据将所述增量数据和所述存量历史数据按照预设业务维度进行数据存储；以所述预设业务维度中任一维度进行数据聚合，将其他数据按照json格式存入elasticsearch中，输出任一维度预设大宽表；将多个业务系统中任一维度的数据根据业务逻辑聚合，并按照数组形式存储至所述任一维度预设大宽表的任一业务字段中，得到所述加工后数据。7.根据权利要求1所述的无侵入式的数据抽取加工方法，其特征在于，对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用，包括：确定存储数据库为elasticsearch；
获取多个业务系统的加工后数据中每个字段数据的聚合计算结果，利用去重标识调用存储原始数据。8.根据权利要求7所述的无侵入式的数据抽取加工方法，其特征在于，利用去重标识调用存储原始数据，包括：将已调用消费数据的主键id存入远程字典服务redis，确定预设有效期；在所述预设有效期内调用数据时将所述redis中的主键id与当前主键id进行对比，若确定不一致，则继续调用，否则停止调用。9.一种无侵入式的数据抽取加工系统，其特征在于，包括：增量提取模块，用于利用增量订阅消费组件提取多个业务系统的增量数据；存量提取模块，用于通过jdbc获取多个业务系统的存量历史数据；加工处理模块，用于对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据；数据调用模块，用于对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用。10.一种电子设备，包括存储器、处理器及存储在所述存储器上并可在所述处理器上运行的计算机程序，其特征在于，所述处理器执行所述程序时实现如权利要求1至8任一项所述无侵入式的数据抽取加工方法。

技术总结
本发明提供一种无侵入式的数据抽取加工方法及系统，属于数据处理技术领域，包括：利用增量订阅消费组件提取多个业务系统的增量数据；通过JDBC获取多个业务系统的存量历史数据；对所述增量数据和所述存量历史数据进行实时加工处理，得到加工后数据；对所述加工后数据进行存储，将存储的加工后数据同步至多个业务系统，以供所述多个业务系统进行实时调用。本发明通过对多种业务系统采用无侵入感知获取业务系统增量及全量数据和实时加工处理数据，实现了实时计算的数据存储、历史数据回溯处理以及支持海量数据快速查询，无需对业务系统进行改造即可进行数据综合管理，具有高实时性和高效率。性和高效率。性和高效率。


技术研发人员：张书庆 田羽 兰翔 汪大磊 吴金林
受保护的技术使用者：武汉众邦银行股份有限公司
技术研发日：2023.06.26
技术公布日：2023/10/11