基于Kafka的实时数据抽取方法及相关设备

摘要

本说明书实施例提供一种基于Kafka的实时数据抽取方法及相关设备。该方法包括：利用SharePlex同步工具实时抽取源系统数据库中的增量数据到Kafka消息集群Topic消息下对应的若干张数据表中；对数据表中的增量数据进行实时抽取；创建表队列，将抽取到的增量数据按照数据表的表名分别放入表队列中；创建入库线程，提取表队列中的数据并解析为嵌入的SQL语句；当SQL语句达到数量阈值或时间阈值时，触发提交逻辑；在SQL Server中创建一个临时内存表，将数据写入所述临时内存表；将临时内存表中的数据写入SQL Server的目标表中。本说明书提供的方法及相关设备，能够实现实时、准确、高效地抽取数据。

说明书

技术领域

本说明书一个或多个实施例涉及数据处理领域，尤其涉及一种基于Kafka的实时数据抽取方法及相关设备。

背景技术

随着互联网技术的发展，越来越多的系统之间需要进行数据的传输与应用，这就需要将源系统数据进行抽取并导入到目的系统。而随着各企业的发展及外部监管的加强，数据的实时性更是迫在眉睫，如管理层的决策要先行，客户的满意度要提高，要快速抓住商机等，这些都对数据实时性提高了要求。

现有的数据抽取技术，大致分为两种：一种是通过甲骨文公司推出的一种数据库方案DataGuard等第三方工具直接链接源系统，产生一份数据，这样会对源系统产生过大压力，造成源系统不稳定，产生的数据也不能满足实际业务需要；另一种是通过在源系统上建立镜像库获取数据，然后通过第三方抽取工具实时抽取数据，这样的方案对源系统压力也非常大，同样不能满足业务实际情况。并且这两种技术最终获取的数据的实时性都不高。

基于此，需要一种能够实现实时从源系统抽取数据的方法。

发明内容

有鉴于此，本说明书一个或多个实施例的目的在于提出一种基于Kafka的实时数据抽取方法及相关设备，以克服现有技术中的不足。

基于上述目的，本说明书一个或多个实施例提供了一种基于Kafka的实时数据抽取方法，包括：

利用数据同步软件SharePlex实时同步抽取源系统数据库Oracle中的增量数据，放入开源流处理平台Kafka的消息集群对应的主题Topic消息下对应的若干张数据表中；

对所述Topic消息下对应的数据表中的增量数据进行实时数据抽取；

按照所述抽取到的增量数据所在的数据表的表名创建对应的若干个表队列，并将所述抽取到的增量数据按照所述表名分别放入所述若干个表队列中；

根据所述若干个表队列，创建对应的若干个入库线程，所述入库线程提取所述表队列中的数据，把所述提取到的数据解析为嵌入的结构化查询语言SQL语句；

触发提交逻辑；

在关系数据库管理系统SQL Server中创建一个临时内存表，将所述被解析为嵌入的SQL语句的数据写入所述临时内存表；

将所述临时内存表中的数据写入所述SQL Server的目标表中。

基于同一发明构思，本说明书一个或多个实施例还提供了一种基于Kafka的实时数据抽取装置，包括：

第一抽取模块，被配置为利用数据同步软件SharePlex实时同步抽取源系统数据库Oracle中的增量数据，放入开源流处理平台Kafka的消息集群对应的主题Topic消息下对应的若干张数据表中；

第二抽取模块，被配置为对所述Topic消息下对应的数据表中的增量数据进行实时数据抽取；

第一创建模块，被配置为按照所述抽取到的增量数据所在的数据表的表名创建对应的若干个表队列，并将所述抽取到的增量数据按照所述表名分别放入所述若干个表队列中；

第二创建模块，被配置为根据所述若干个表队列，创建对应的若干个入库线程，所述入库线程提取所述表队列中的数据，把所述提取到的数据解析为嵌入的结构化查询语言SQL语句；

触发模块，被配置为触发提交逻辑；

第一写入模块，被配置为在关系数据库管理系统SQL Server中创建一个临时内存表，将所述被解析为嵌入的SQL语句的数据写入所述临时内存表；

第二写入模块，被配置为将所述临时内存表中的数据写入所述SQL Server的目标表中。

基于同一发明构思，本说明书一个或多个实施例还提供了一种电子设备，包括存储器、处理器及存储在存储器上并可在处理器上运行的计算机程序，所述处理器执行所述程序时实现如上任意一项所述的方法。

基于同一发明构思，本说明书一个或多个实施例还提供了一种非暂态计算机可读存储介质，其中，所述非暂态计算机可读存储介质存储计算机指令，所述计算机指令用于使所述计算机执行如上任意一项所述的方法。

从上面所述可以看出，本说明书一个或多个实施例的基于Kafka的实时数据抽取方法及相关设备，利用数据同步软件SharePlex与开源流处理平台Kafka抽取数据、并根据抽取到的数据动态地创建表队列与入库线程，能够实现从源系统数据库中实时、准确、高效地抽取数据放入目标表中，并且能够减少源系统的压力。

附图说明

为了更清楚地说明本说明书一个或多个实施例或现有技术中的技术方案，下面将对实施例或现有技术描述中所需要使用的附图作简单地介绍，显而易见地，下面描述中的附图仅仅是本说明书一个或多个实施例，对于本领域普通技术人员来讲，在不付出创造性劳动的前提下，还可以根据这些附图获得其他的附图。

图1为本说明书一个或多个实施例的基于Kafka的实时数据抽取方法流程图；

图2为本说明书一个或多个实施例的基于Kafka的实时数据抽取方法具体流程图；

图3为本说明书一个或多个实施例中的限流方法流程图；

图4为本说明书一个或多个实施例中的健康检查方法流程图；

图5为本说明书一个或多个实施例的基于Kafka的实时数据抽取装置结构示意图；

图6为本说明书一个或多个实施例的电子设备结构示意图。

具体实施方式

为使本公开的目的、技术方案和优点更加清楚明白，以下结合具体实施例，并参照附图，对本公开进一步详细说明。

需要说明的是，除非另外定义，本说明书一个或多个实施例使用的技术术语或者科学术语应当为本公开所属领域内具有一般技能的人士所理解的通常意义。本说明书一个或多个实施例中使用的“第一”、“第二”以及类似的词语并不表示任何顺序、数量或者重要性，而只是用来区分不同的组成部分。“包括”或者“包含”等类似的词语意指出现该词前面的元件或者物件涵盖出现在该词后面列举的元件或者物件及其等同，而不排除其他元件或者物件。

如背景技术部分所述，现有的数据抽取方法实时性不足，还难以满足企业需求。申请人在实现本公开技术方案的过程中发现，现有的数据抽取方法还存在以下不足：

(1)没有限流措施：当业务高峰来临，源系统产生大量数据时，数据一下子涌入，抽取系统处理效率低下，导致大量积压，占用大量输入输出(IO)，抽取速度急剧下降；

(2)没有熔断措施：当网络延迟或者异常数据等原因导致数据写入失败，会造成数据阻塞，致使数据法正常写入目标库。

有鉴于此，本说明书一个或多个实施例提供了一种基于Kafka的实时数据抽取方法，参考图1，包括以下步骤：

步骤S101、利用数据同步软件SharePlex实时同步抽取源系统数据库Oracle中的增量数据，放入开源流处理平台Kafka的消息集群对应的主题Topic消息下对应的若干张数据表中；

步骤S102、对所述Topic消息下对应的数据表中的增量数据进行实时数据抽取；

步骤S103、按照所述抽取到的增量数据所在的数据表的表名创建对应的若干个表队列，并将所述抽取到的增量数据按照所述表名分别放入所述若干个表队列中；

步骤S104、根据所述若干个表队列，创建对应的若干个入库线程，所述入库线程提取所述表队列中的数据，把所述提取到的数据解析为嵌入的结构化查询语言SQL语句；

步骤S105、触发提交逻辑；

步骤S106、在关系数据库管理系统SQL Server中创建一个临时内存表，将所述被解析为嵌入的SQL语句的数据写入所述临时内存表；

步骤S107、将所述临时内存表中的数据写入所述SQL Server的目标表中。

可见，本说明书一个或多个实施例的基于Kafka的实时数据抽取方法，利用数据同步软件SharePlex与开源流处理平台Kafka抽取数据、并根据抽取到的数据动态地创建表队列与入库线程，能够实现从源系统数据库中实时、准确、高效地抽取数据放入目标表中，并且能够减少源系统的压力。

以下，通过具体的实施例来详细说明本说明书一个或多个实施例的技术方案。

参考图2，为本说明书一个实施例的基于Kafka的实时数据抽取方法具体流程图，包括以下步骤：

步骤S201、利用数据同步软件SharePlex实时同步抽取源系统数据库Oracle中的增量数据，放入开源流处理平台Kafka的消息集群对应的主题Topic消息下对应的若干张数据表中。

本步骤中，源系统数据库Oracle会实时记录增量数据并保存到在线日志文件中，可利用数据同步软件SharePlex通过所述在线日志文件实时同步抽取所述增量数据；然后，将所述增量数据放入开源流处理平台Kafka的消息集群对应的主题Topic消息下对应的若干张数据表中。其中，Topic消息下的若干张数据表用来存放源系统数据库Oracle中相应表的增量数据。

步骤S202、启动实时数据抽取程序，通过消费线程Consumer对所述Topic消息下对应的若干张数据表中的增量数据进行实时数据抽取。

本步骤中，启动实时数据抽取程序后，所述实时数据抽取程序会加载其配置库中元数据信息(Topic信息、Topic消息下对应的数据表的表名、Topic消息下对应的数据表的表结构等)，根据启动时传入的Topic参数建立起一个与Kafka的数据传输通道，然后实时数据抽取程序启动会消费线程Consumer，消费线程根据此数据传输通道对Topic消息下对应的若干张数据表中的增量数据进行实时数据抽取。此时，抽取到的数据为JSON格式，所述实时数据抽取程序会把抽取到的JSON格式的数据解析并且封装为相应的Java对象，所述Java对象能够被转换为Java格式的数据。

另外，当Kafka的消息集群出现故障时，会导致无法继续抽取数据，此时，实时数据抽取程序定时监测Kafka的消息集群的健康恢复状况，一旦Kafka的消息集群恢复健康，则继续对其对应的Topic消息下对应的若干张数据表中的增量数据进行实时数据抽取，无需人工干预。

步骤S203、定时器定时(每隔15秒或20秒)记录消费线程的offset信息并写入Topic状态信息表。

本步骤中，当有源系统数据库有新增表，需要对所述新增表进行数据增量实时抽取时，为了避免Kafka的offset信息过多的回溯导致Topic消息下对应的其他数据表的增量数据不再更新，实时数据抽取程序会启动一个次消费线程，这个次消费线程只对新增表的数据进行抽取，定时器定时(每隔15秒或20秒)记录消费线程和次消费线程的offset信息并写入Topic状态信息表，当次消费线程的offset信息追平消费线程时，实时数据抽取程序会把新增表中的增量数据纳入主消费线程中，并关闭次消费线程。

步骤S204、实时数据抽取程序根据消费线程抽取到的Topic消息下对应的若干数据张表的增量数据所在的数据表的表名，到资源池拿取资源，动态地创建对应的若干个表队列Queue；并根据所述若干个表队列，到资源池拿取资源，动态地创建与表队列一一对应的若干个入库线程。

步骤S205、将所述抽取到的增量数据按照所述数据表的表名分别放入所述若干个表队列中。

本步骤中，实时数据抽取程序将所述抽取到的增量数据按照所述数据表的表名分别放入所述若干个表队列中。但是，当长时间没有Topic消息下对应的某个数据表的增量数据被抽取到时，实时数据抽取程序会回收对应存放该数据表的增量数据表队列资源，同时，此表队列对应的入库线程也会被回收。此处，实时数据抽取程序的程序逻辑可运用Java的锁机制，保证表队列和入库线程的重复开启。这样做的目的是为了更好的优化实时数据抽取程序使用的资源。

步骤S206、定时器定时(每隔20秒或30秒)记录每个表队列的积压量到队列长度表，查看还有多少未处理的数据。

步骤S207、入库线程提取所述表队列中的数据，把所述提取到的数据解析为嵌入的结构化查询语言SQL语句。

本步骤中，所述入库线程提取到的数据数据为Java对象，然后把Java对象解析为嵌入insert的结构化查询语言SQL语句；接着入库线程会将SQL语句批量装配成SQLBatch放入SQLBatch提交线程池中。

步骤S208、当所述SQLBatch提交线程池中的SQL语句达到数量阈值或时间阈值时，触发提交逻辑。

本步骤中，提交逻辑触发的一瞬间，实时数据抽取程序会在关系数据库管理系统SQL Server中创建一个临时内存表，SQLBatch提交线程池中的SQL语句会先写入临时内存表。

步骤S209、将所述临时内存表中的数据写入所述SQL Server的目标表中。

本步骤中，数据在写入临时内存表后，实时数据抽取程序会调用一个创建好的存储过程，该存储过程执行流程是：首先通过主键字段关联目标表，然后删除目标表关联到的数据条目，最后把临时内存表的数据写入目标表。这么做是为了达到同主键数据更新、新数据写入的目的，也就是源系统数据库Oracle的新增和更新数据的操作。对于源系统数据库Oracle的删除数据操作，目标表不进行物理删除，而是打上删除表示位，相当于更新。

进一步的，本实施例的实时数据抽取程序还包括：限流设计，参考图3，为本说明书一个实施例中的限流方法流程图，包括以下步骤：

步骤S301、实时数据抽取程序向Kafka发出请求消息。

步骤S302、实时数据抽取程序接收Kafka回复的批量消息。

步骤S303、判断消息是否跳过。

本步骤中，Kafka回复的批量消息中可能会包含不需要的非业务处理数据，对于不需要处理的消息实时数据抽取程序会自动跳过。

步骤S304、若不跳过，则对消息进行加工处理。

步骤S305、全部消息处理完成后计算耗时。

步骤S306、当耗时超过15秒时，继续向Kafka发出下次请求消息；当向Kafka发出请求消息的频次过高时，则自动休眠25秒后再继续向Kafka发出下次请求消息。

实时数据抽取程序的数据源来自于Kafka，当出现大批量不需要加工的数据时，实时数据抽取程序的请求频次会升高，导致服务器带宽超负荷。因此采用了限流设计，当处理消息的时间耗时过短时，实时数据抽取程序会延缓发出下次请求。

进一步的，本实施例的实时数据抽取程序还包括：熔断设计。具体的，Topic消息下对应的若干张数据表的增量数据会被放入相应的表队列中，相应的入库线程会从表队列中取出数据并把数据解析为SQL语句，然后批量装配成SQLBatch放入提交线程池中。如果出现SQL语句写入目标表异常，提交线程池会重试3次提交，如果SQL语句仍无法成功写入目标库，实时数据抽取程序会关闭该数据表的所有相关表队列和入库线程，避免Kafka消费模块由于表队列长度过满导致的阻塞，致使其它表的数据无法正常写入目标表。

进一步的，本实施例的实时数据抽取程序还包括：健康检查设计。具体的，实时数据抽取程序启动后，会启动一个心跳定时器对消费线程的运行状态进行健康检查。参考图4，为本说明书一个实施例中的健康检查方法流程图，包括以下步骤：

S401、心跳定时器每隔60秒对消费线程的运行状态进行检查；

S402、判断配置库开关状态；

S403、若配置库处于关闭状态，则无需开启检查，关闭流程；

S404、若配置库处于开启状态，则开启检查，判断消费线程是否正常；

S405、若消费线程正常运行，则将心跳信息写入配置库；

S406、若消费线程状态异常，则重启消费线程。

可见，本说明书实施例提供的基于Kafka的实时数据抽取方法，利用数据同步软件SharePlex与开源流处理平台Kafka，能够从源系统数据库中实时地抽取数据并且减少源系统的压力。通过创建表队列、入库线程以及提交线程池，将抽取到的数据准确、高效地存入目标表中。另外，采用限流设计，能够减轻服务器的负荷，避免数据大量积压，提高数据抽取效率。采用熔断设计，能够避免数据阻塞、数据无法正常存入目标表等问题。

可以理解，该方法可以通过任何具有计算、处理能力的装置、设备、平台、设备集群来执行。

需要说明的是，本说明书一个或多个实施例的方法可以由单个设备执行，例如一台计算机或服务器等。本实施例的方法也可以应用于分布式场景下，由多台设备相互配合来完成。在这种分布式场景的情况下，这多台设备中的一台设备可以只执行本说明书一个或多个实施例的方法中的某一个或多个步骤，这多台设备相互之间会进行交互以完成所述的方法。

需要说明的是，上述对本说明书特定实施例进行了描述。其它实施例在所附权利要求书的范围内。在一些情况下，在权利要求书中记载的动作或步骤可以按照不同于实施例中的顺序来执行并且仍然可以实现期望的结果。另外，在附图中描绘的过程不一定要求示出的特定顺序或者连续顺序才能实现期望的结果。在某些实施方式中，多任务处理和并行处理也是可以的或者可能是有利的。

基于同一发明构思，本说明书一个或多个实施例还提供了一种基于Kafka的实时数据抽取装置。参考图5，所述的基于Kafka的实时数据抽取装置，包括：

第一抽取模块501，被配置为利用数据同步软件SharePlex实时同步抽取源系统数据库Oracle中的增量数据，放入开源流处理平台Kafka的消息集群对应的主题Topic消息下对应的若干张数据表中；

第二抽取模块502，被配置为对所述Topic消息下对应的数据表中的增量数据进行实时数据抽取；

第一创建模块503，被配置为按照所述抽取到的增量数据所在的数据表的表名创建对应的若干个表队列，并将所述抽取到的增量数据按照所述表名分别放入所述若干个表队列中；

第二创建模块504，被配置为根据所述若干个表队列，创建对应的若干个入库线程，所述入库线程提取所述表队列中的数据，把所述提取到的数据解析为嵌入的结构化查询语言SQL语句；

触发模块505，被配置为触发提交逻辑；

第一写入模块506，被配置为在关系数据库管理系统SQLServer中创建一个临时内存表，将所述被解析为嵌入的SQL语句的数据写入所述临时内存表；

第二写入模块507，被配置为将所述临时内存表中的数据写入所述SQL Server的目标表中。

作为一个可选的实施例，当所述Kafka的消息集群出现故障时，定时监测所述Kafka的消息集群的健康恢复状况，当所述Kafka的消息集群恢复健康，则继续对其对应的Topic消息下对应的数据表中的增量数据进行实时数据抽取。

作为一个可选的实施例，在对所述Topic消息下对应的数据表中的增量数据进行实时数据抽取之后，还包括：将所述进行实时数据抽取得到的增量数据进行解析并封装为Java对象，所述增量数据为JSON格式的数据。

作为一个可选的实施例，所述触发提交逻辑的条件为：当所述SQL语句达到数量阈值或时间阈值。

作为一个可选的实施例，所述利用数据同步软件SharePlex实时同步抽取源系统数据库Oracle中的增量数据，具体被配置为源系统数据库Oracle实时记录增量数据并保存到在线日志文件中，利用数据同步软件SharePlex的服务器通过所述在线日志文件实时同步抽取所述增量数据。

作为一个可选的实施例，所述第二抽取模块502，具体被配置为加载配置库中元数据信息，所述元数据包括Topic信息、Topic消息下对应的数据表的表名、Topic消息下对应的数据表的表结构；根据Topic参数建立一个与开源流处理平台kafka的数据传输通道；根据数据传输通道对所述Topic消息下对应的数据表中的增量数据进行实时数据抽取。

作为一个可选的实施例，所述第二写入模块507，具体被配置为通过主键字段关联目标表；删除所述目标表关联到的数据条目；将所述临时内存表中的数据写入目标表。

为了描述的方便，描述以上装置时以功能分为各种模块分别描述。当然，在实施本说明书一个或多个实施例时可以把各模块的功能在同一个或多个软件和/或硬件中实现。

上述实施例的装置用于实现前述实施例中相应的基于Kafka的实时数据抽取方法，并且具有相应的方法实施例的有益效果，在此不再赘述。

基于同一发明构思，与上述任意实施例方法相对应的，本说明书一个或多个实施例还提供了一种电子设备，包括存储器、处理器及存储在所述存储器上并可在所述处理器上运行的计算机程序，所述处理器执行所述计算机程序时实现如上任意一实施例所述的基于Kafka的实时数据抽取方法。

图6示出了本实施例所提供的一种更为具体的电子设备硬件结构示意图，该设备可以包括：处理器1010、存储器1020、输入/输出接口1030、通信接口1040和总线1050。其中处理器1010、存储器1020、输入/输出接口1030和通信接口1040通过总线1050实现彼此之间在设备内部的通信连接。

处理器1010可以采用通用的CPU(Central Processing Unit，中央处理器)、微处理器、应用专用集成电路(Application Specific Integrated Circuit，ASIC)、或者一个或多个集成电路等方式实现，用于执行相关程序，以实现本说明书实施例所提供的技术方案。

存储器1020可以采用ROM(Read Only Memory，只读存储器)、RAM(RandomAccessMemory，随机存取存储器)、静态存储设备，动态存储设备等形式实现。存储器1020可以存储操作系统和其他应用程序，在通过软件或者固件来实现本说明书实施例所提供的技术方案时，相关的程序代码保存在存储器1020中，并由处理器1010来调用执行。

输入/输出接口1030用于连接输入/输出模块，以实现信息输入及输出。输入输出/模块可以作为组件配置在设备中(图中未示出)，也可以外接于设备以提供相应功能。其中输入设备可以包括键盘、鼠标、触摸屏、麦克风、各类传感器等，输出设备可以包括显示器、扬声器、振动器、指示灯等。

通信接口1040用于连接通信模块(图中未示出)，以实现本设备与其他设备的通信交互。其中通信模块可以通过有线方式(例如USB、网线等)实现通信，也可以通过无线方式(例如移动网络、WIFI、蓝牙等)实现通信。

总线1050包括一通路，在设备的各个组件(例如处理器1010、存储器1020、输入/输出接口1030和通信接口1040)之间传输信息。

需要说明的是，尽管上述设备仅示出了处理器1010、存储器1020、输入/输出接口1030、通信接口1040以及总线1050，但是在具体实施过程中，该设备还可以包括实现正常运行所必需的其他组件。此外，本领域的技术人员可以理解的是，上述设备中也可以仅包含实现本说明书实施例方案所必需的组件，而不必包含图中所示的全部组件。

上述实施例的电子设备用于实现前述任一实施例中相应的基于Kafka的实时数据抽取方法，并且具有相应的方法实施例的有益效果，在此不再赘述。

基于同一发明构思，与上述任意实施例方法相对应的，本说明书一个或多个实施例还提供了一种非暂态计算机可读存储介质，所述非暂态计算机可读存储介质存储计算机指令，所述计算机指令用于使所述计算机执行如上任一实施例所述的基于Kafka的实时数据抽取方法。

本实施例的计算机可读介质包括永久性和非永久性、可移动和非可移动媒体可以由任何方法或技术来实现信息存储。信息可以是计算机可读指令、数据结构、程序的模块或其他数据。计算机的存储介质的例子包括，但不限于相变内存(PRAM)、静态随机存取存储器(SRAM)、动态随机存取存储器(DRAM)、其他类型的随机存取存储器(RAM)、只读存储器(ROM)、电可擦除可编程只读存储器(EEPROM)、快闪记忆体或其他内存技术、只读光盘只读存储器(CD-ROM)、数字多功能光盘(DVD)或其他光学存储、磁盒式磁带，磁带磁磁盘存储或其他磁性存储设备或任何其他非传输介质，可用于存储可以被计算设备访问的信息。

上述实施例的存储介质存储的计算机指令用于使所述计算机执行如上任一实施例所述的基于Kafka的实时数据抽取方法，并且具有相应的方法实施例的有益效果，在此不再赘述。

所属领域的普通技术人员应当理解：以上任何实施例的讨论仅为示例性的，并非旨在暗示本公开的范围(包括权利要求)被限于这些例子；在本公开的思路下，以上实施例或者不同实施例中的技术特征之间也可以进行组合，步骤可以以任意顺序实现，并存在如上所述的本说明书一个或多个实施例的不同方面的许多其它变化，为了简明它们没有在细节中提供。

尽管已经结合了本公开的具体实施例对本公开进行了描述，但是根据前面的描述，这些实施例的很多替换、修改和变型对本领域普通技术人员来说将是显而易见的。例如，其它存储器架构(例如，动态RAM(DRAM))可以使用所讨论的实施例。

本说明书一个或多个实施例旨在涵盖落入所附权利要求的宽泛范围之内的所有这样的替换、修改和变型。因此，凡在本说明书一个或多个实施例的精神和原则之内，所做的任何省略、修改、等同替换、改进等，均应包含在本公开的保护范围之内。