CN114048217B 增量数据的同步方法和装置、电子设备和存储介质 （微民保险代理有限公司）

本申请提供了一种增量数据的同步方法和在确定源端的目标业务系统内的目标子业务被利用流式处理引擎提取主题分区中的操作记录并将操作记录排序队列对应的增量数据发送至2在确定源端的目标业务系统内的目标子业务被更新的情况下将所述操作记录按照所述关联信息的分类分别发送至分布式消息队列的各个主题分述时间戳的先后次序对所述消费后的所述操作记录进行排序后生成的队列；获取所述目标业务系统内的第一数据库，其中，所述第一数据库将所述可读取数据转换为二进制日志文件，其中，所述二进制根据解析工具解析所述二进制日志文件，得到所述目标子业务被更对所述操作记录排序队列对应的所述增量数据进行格式的转换，得将所述转换格式后的增量数据发送至所述目标端，其中，所述目标端所述流式处理引擎根据预设消费方案对读取的所述操作记录排序队列进行消费，其3获取所述操作记录排序队列内的各个所述操作将所述检查机制指示错误的所述操作记录从所述操作记录在利用所述流式处理引擎消费所述操作记录的情况下，第一获取单元，用于在确定源端的目标业务系统内的目标子业得到单元，用于利用流式处理引擎提取所述主题分区中的所述操作列是依据所述时间戳的先后次序对所述消费后的所述操作记录进行排序后生成的队列；第二发送单元，用于利用所述流式处理引擎读取所述所述处理器，用于通过运行所述存储器上所存储的所述计算机程序来执行权利要求14处理器执行时实现权利要求1-7任一项所述的增量数据5[0002]目前，存在一些源端数据库和目标端数据库中存储的数决相关技术中存在源端数据和目标端数据的增量数[0008]将所述操作记录按照所述关联信息的分类分别发送至分布式消息队列的各个主先后次序对所述消费后的所述操作记录进行排序后生成的队列；[0012]第一获取单元，用于在确定源端的目标业务系统内的目标子业务被更新的情况6[0014]得到单元，用于利用流式处理引擎提取所述主题分区中的所述操作记录进行消依据所述时间戳的先后次序对所述消费后的所述操作记录进行排序后生成的队算机程序来执行上述任一实施例中的增量数[0018]根据本申请实施例的又一个方面，还提供了一种计算机目标业务系统内包含多个子业务，操作记录用于表征增量数据产生时对应的多个关联信些操作记录记录在第二数据库内，利用第二数据库作为增量同步操作记录的中转存储介7员在没有做出创造性劳动前提下所获得的所有其他实施例，都应当属于本申请保护的范8的增量数据所对应的操作记录，该操作记录内存储有多个与该增量数据相关的关联信息，景分类将操作记录分别发送至kafka的各个主目标业务系统内包含多个子业务，操作记录用于表征增量数据产生时对应的多个关联信9处理引擎提取主题分区中的操作记录进行消费，并将消费后的操作记录写入第二数据库[0053]然后通过数据库操作记录解析工具(比如xxxx的canal开源解析工具)解析Binlog[0056]{"data":[{"id":"29","order_id":"291029","amount":"281727.39","create_time":"2020-09-2503:17:59"}],"database":"testbase","es":1600975079000,"id":4,"isDdl":false,"mysqlType":{"id":"BIGINT","order_id":"VARCHAR(64)","amount":"DECIMAL(10,2)","create_time":"DATETIME"},"old":[{"amount":"666.0"}],"pkNames":["id"],"sql":"","sqlType":{"id":-5,"order_id":12,"amount":3,"create_time":93},"table":"order","ts":160097517900[0059]作为一种可选实施例，利用流式处理引擎提取主题分区按照预设格式将Flink消费后的操作记录写入第二数据库的预设格式对应的预设单元内，戳作为版本区分方式，Hbase支持的表结构格式的示例如下：(rowkey,handle_msg,datasource,table,handle_type,handl[0070](1,{"id":"1","order_id":"10086","amount":"10087.0","create_time":"2020-03-0205:12:49"},"test","order","UP[0073]在本申请实施例中，Flink消费的数据都带有操作记录时间戳，将该时间戳作为Hbase单个单元记录的版本号，对于由于网络延时或者是多分区下消费速率不一致导致的[0074]作为一种可选实施例，在将操作记录排序队列对应的增量数据发送至目标端之的数据库支持格式，本申请实施例将操作记录排序队列对应的增量数据进行格式的转换，根据数据写入操作指令，对mysql数据库中的增量数据进行格式更改，以适应Hive数据仓设消费方案用于指示流式处理引擎对操作记录排序队列内的各个操作记录执行有且仅有号相同的操作记录或者识别码的编号有误的操作记录从操作记即当前的最新的版本号，然后将预设单元内存放的所有版本号与该第一版本号进行数值大就是时间戳，而时间戳又是由当前时刻转换来的，所以如果所有版本号中存在数值小于第一版本号的第二版本号，则证明该操作记录已消费完毕，这些操作记录的数据就相当于已第一版本号之外的其他任意一个版本号。[0098]具体地，mysql实例经过合理配置后，业务系统对数据库的操作转存为二进制的[0100]具体可以通过数据库操作记录解析工具(比如canal开源解析工具)解析mysql的中的状态重启，而不丢失数据。Flink将解析出操作记录的时间戳及其他关键信息写入时或者是多partition下消费速率不一致导致的乱序数据，可由Hbase基于timestamp的多号，即删除过期的操作记录，输入不同的目标端，例如Hive数据仓库、ES搜索服务引擎、mysql实例。支持，可以依次解析Hbase中存储排序后的json数据到Flinksql中。并由Flink对接写入格式(例如ES中的一个doc、Hbase中的一条put)，与此同时将偏移量保存在状态信息法的增量数据的同步装置。图3是根据本申请实施例的一种可选的增量数据的同步装置的[0114]第一获取单元301，用于在确定源端的目标业务系统内的目标子业务被更新的情时间戳的先后次序对消费后的操作记录进行排序后生成些操作记录记录在第二数据库内，利用第二数据库作为增量同步操作记录的中转存储介[0140]第一删除单元，用于将检查机制指示错误的操作记录从操作记录排序队列中删[0154]可选地，在本实施例中，上述的通信总线可以是PCI(PeripheralComponentInterconnect，外设部件互连标准)总线、或EISA(ExtendedIndustryStandard[0158]上述处理器可以是通用处理器，可以包含但不限于：CPU(CentralProcessingUnit，中央处理器)、NP(NetworkProcessor，网络处理器)等；还可以是DSP(DigitalSignalProcessing，数字信号处理器)、ASIC(ApplicationSpecificIntegrated[0162]本领域普通技术人员可以理解上述实施例的各种方法中的全部或部分步骤是可[0172]根据本申请实施例的又一个方面