本发明涉及数据处理，尤其涉及一种数据写入处理方法、装置、电子设备及存储介质。
背景技术：
：1、目前市面基于flink写入数据基本采用doris提供的已经封装好的(flink dorisconnector)进行写入；2、在使用sql版本(flink doris sink)写入数据时存在无法单独设定并行度(parallelism)的问题，但是为了提高doris吞吐量，必须将程序总并行度设置较高，导致写入频率过快，doris tablet version过多，doris表聚合(compaction)速度无法满足，最终写入doris频繁失败，结果并没有提高写入doris的吞吐量；在使用datastream版本(flinkdoris sink)写入数据时存在无法批量数据写入的问题，这就导致即便此种方式能够单独设定并行度降低doris的写入频率，也无法有效提高写入doris的吞吐量。技术实现思路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、基于连接信息和配置文件信息，采用http协议向目标数据库发起写入请求，并通过基于http协议传输的导入方式将数据队列中的所有单条数据批量写入目标数据库；进而接收目标数据库返回的写入状态；30、当写入状态中存在异常写入信息，将异常写入信息对应的单条数据回滚到数据队列中。31、另一方面，本发明实施例提供了一种数据写入处理装置，包括：32、第一模块，用于基于第一启动方法开启预设定时器；33、第二模块，用于响应于预设定时器的启动动作，调用第二启动方法初始化批量数据条数和数据队列以及建立源数据库与预设接口组件的传输链接；34、第三模块，用于基于传输链接，通过预设接口组件调用数据读取方法持续从源数据库获取待写入数据；其中，待写入数据为流数据，流数据包括连续的多个单条数据；35、第四模块，用于将流数据中的每个单条数据依次存入数据队列；36、第五模块，用于当数据队列中的数据元素数量达到批量数据条数，将数据队列中的所有单条数据批量写入目标数据库；调用关闭方法关闭传输链接。37、可选地，执行第五模块后，装置还包括：38、第六模块，用于返回执行第一模块，以循环执行前述的数据写入处理装置中的各个模块持续进行数据写入处理；39、可选地，第二模块具体用于：40、获取单条数据的数据大小和数据队列的数据大小总限；41、基于数据大小总限与数据大小的比值确定第一条数；42、获取预设接口组件的自定义并行度；43、基于流程总并行度和流程参数的积与自定义并行度的比值确定第二条数；其中，流程总并行度和流程参数基于目标数据库的性能特征确定；44、根据第一条数和第二条数中的最小值作为批量数据条数。45、可选地，当待写入数据不为流数据，装置还包括：46、第七模块，用于获取待写入数据的表数据格式；47、第八模块，用于基于待写入数据的表数据格式，调用预设转换方法将待写入数据转换为流数据。48、可选地，执行第四模块前，装置还包括：49、第九模块，用于将流数据中的每个单条数据处理为统一的预设数据格式；50、第十模块，用于当存在单条数据无法处理为预设数据格式，将单条数据标记为异常数据；51、第十一模块，用于基于预设异常数据处理机制对异常数据进行异常数据处理。52、可选地，第四模块具体用于：53、将流数据中首个待存入数据队列中的单条数据作为目标数据；54、将目标数据存入数据队列，获取数据队列中的数据元素数量；55、当数据元素数量小于批量数据条数，将流数据中目标数据的下一个单条数据作为目标数据，然后返回执行将目标数据存入数据队列的步骤，直至数据队列中的数据元素数量达到批量数据条数。56、可选地，第五模块具体用于：57、获取目标数据库的连接信息和配置文件信息；58、基于连接信息和配置文件信息，采用http协议向目标数据库发起写入请求，并通过基于http协议传输的导入方式将数据队列中的所有单条数据批量写入目标数据库；进而接收目标数据库返回的写入状态；59、当写入状态中存在异常写入信息，将异常写入信息对应的单条数据回滚到数据队列中。60、另一方面，本发明实施例提供了一种电子设备，包括：处理器以及存储器；存储器用于存储程序；处理器执行程序实现上述数据写入处理方法。61、另一方面，本发明实施例提供了一种计算机存储介质，其中存储有处理器可执行的程序，处理器可执行的程序在由处理器执行时用于实现上述数据写入处理方法。62、本发明实施例通过基于第一启动方法开启预设定时器；响应于预设定时器的启动动作，调用第二启动方法初始化批量数据条数和数据队列以及建立源数据库与预设接口组件的传输链接；基于传输链接，通过预设接口组件调用数据读取方法持续从源数据库获取待写入数据；其中，待写入数据为流数据，流数据包括连续的多个单条数据；将流数据中的每个单条数据依次存入数据队列；当数据队列中的数据元素数量达到批量数据条数，将数据队列中的所有单条数据批量写入目标数据库；调用关闭方法关闭传输链接。本发明实施例通过预设定时器来对海量数据进行扫尾并写入目标数据库中，以此提高海量数据写入目标数据库的吞吐量，达到前处理高并行度，最后写入低并行度下依旧能够稳定高吞吐量去写入数据的效果。本发明实施例能够高效进行数据写入处理。当前第1页12
技术特征：1.一种数据写入处理方法，其特征在于，包括以下步骤：
2.根据权利要求1所述的数据写入处理方法，其特征在于，所述调用关闭方法关闭所述传输链接这一步骤之后，所述方法还包括以下步骤：
3.根据权利要求1所述的数据写入处理方法，其特征在于，所述初始化批量数据条数，包括以下步骤：
4.根据权利要求1所述的数据写入处理方法，其特征在于，当所述待写入数据不为所述流数据，所述方法还包括以下步骤：
5.根据权利要求1所述的数据写入处理方法，其特征在于，所述将所述流数据中的每个所述单条数据依次存入所述数据队列这一步骤之前，所述方法还包括以下步骤：
6.根据权利要求1所述的数据写入处理方法，其特征在于，所述将所述流数据中的每个所述单条数据依次存入所述数据队列，包括以下步骤：
7.根据权利要求1所述的数据写入处理方法，其特征在于，所述将所述数据队列中的所有所述单条数据批量写入目标数据库，包括以下步骤：
8.一种数据写入处理装置，其特征在于，包括：
9.一种电子设备，其特征在于，包括处理器以及存储器；
10.一种计算机存储介质，其中存储有处理器可执行的程序，其特征在于，所述处理器可执行的程序在由所述处理器执行时用于实现如权利要求1至7任一项所述的方法。
技术总结本发明公开了一种数据写入处理方法、装置、电子设备及存储介质，方法包括：基于第一启动方法开启预设定时器；响应于预设定时器的启动动作，调用第二启动方法初始化批量数据条数和数据队列以及建立源数据库与预设接口组件的传输链接；基于传输链接，通过预设接口组件调用数据读取方法持续从源数据库获取待写入数据；待写入数据为流数据；将流数据中的每个单条数据依次存入数据队列；当数据队列中的数据元素数量达到批量数据条数，将数据队列中的所有单条数据批量写入目标数据库；调用关闭方法关闭传输链接。本发明能够高效进行数据写入处理，可广泛应用于数据处理技术领域。
技术研发人员：张占业
受保护的技术使用者：中电信人工智能科技（北京）有限公司
技术研发日：技术公布日：2024/10/31
