多源异构数据仓库之间的通信方法、装置、设备及介质与流程

本发明涉及数据处理，特别涉及多源异构数据仓库之间的通信方法、装置、设备及介质。

背景技术：

1、在金融等数据密集型业务场景中，数据仓库（含离线数据仓库与实时数据仓库）需要向数十甚至上百个外围业务系统持续下发数据，这些外围业务系统涵盖app（application，应用软件）、核算、征信等诸多方向。上述外围业务系统在接入协议、存储介质、数据格式、写入方式上千差万别，可能是关系型数据库、nosql（not only sql，非关系型数据库）数据库、消息队列、搜索引擎、图数据库、文件系统等十余种不同介质，因此数据仓库对外下发数据时面临严峻的异构适配挑战。相关技术中，数据仓库向外围业务系统下发数据主要存在以下几类实现方式。第一类为点对点直连方案，即业务系统通过jdbc（javadatabase connectivity，java数据库连接）、sqoop脚本、shell任务等方式直接连接数据仓库的hive、hdfs（hadoop distributed file system，一种分布式文件系统）、impala进行数据拉取。第二类为通过kettle、datax、informatica等工具，以图形化配置的方式定义源端到目标端的数据流。第三类为基于消息队列的数据分发方案，即将数据仓库的数据全量推入kafka，由下游各系统各自订阅消费。第四类为手工脚本加定时任务方案，由运维人员维护脚本完成数据的导出、传输、加载。

2、然而，上述相关技术至少存在如下缺陷。其一，集群与下游耦合度高，每接入一个新的下游系统或更换一个数据仓库集群，都需要重写适配代码，扩展性差、研发成本高。其二，不同数据仓库集群的执行引擎差异大，例如impala与spark sql在语法、临时表创建机制、hdfs路径规范上均不相同，业务方需要分别适配，学习成本高、易出错。其三，缺乏统一的权限管控，数据仓库直接暴露给下游系统，无法做到库、表、字段、系统级别的细粒度权限校验，安全风险高。其四，数据一致性难以保证，在多线程批处理、网络传输、目标系统写入故障等场景下，无法精准统计行数、定位失败任务，存在数据错位风险，问题排查耗时长。其五，吞吐能力受限，传统etl（extraction transformation loading，提取转换加载）工具采用单点处理，在tb（terabyte，太字节）级数据下发时性能瓶颈明显。其六，缺乏统一的监控与告警，散点式直连导致任务状态不可见、异常无法及时通知，运维效率低。

3、因此，如何在一套系统中统一对接多个异构数据仓库集群、以一套标准化接口支持多种异构下游目标系统的写入、在保证数据一致性与可追溯性的同时实现高吞吐数据下发、并对数据下发实施统一的权限管控与异常告警，是本领域亟待解决的技术问题。

技术实现思路

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、将添加所述正则替换函数和所述类型转换函数后的查询语句，作为更新后的查询语句写入所述创建临时表的结构化查询语言语句中；

30、基于当前临时表的表名生成新的临时表名，基于所述新的临时表名重新执行写入有所述更新后的查询语句的所述创建临时表的结构化查询语言语句，以得到新的临时表，将所述新的临时表确定为当前临时表。

31、可选的，当前临时表的表名中包含所述数据传输请求的请求标识；

32、相应地，所述通过所述分布式文件系统客户端将当前临时表写入分布式文件系统上临时路径下的文件之后，还包括：

33、按照预设调度周期检查所述临时路径下各当前临时表的存续时间；

34、将所述存续时间超过预设时间阈值的当前临时表对应的数据文件及元数据删除。

35、可选的，所述根据所述数据仓库集群标识查询预设集群路由表之前，还包括：

36、校验所述数据传输请求携带的密钥与系统名之间的对应关系是否成立；

37、校验由所述系统名与目标源表构成的组合是否属于预设的系统-表权限白名单；所述目标源表为所述数据源参数中包含的用于指示待传输数据来源的表；

38、若均校验成功，则触发所述根据所述数据仓库集群标识查询预设集群路由表的操作；

39、否则拒绝所述数据传输请求，并写入审计日志。

40、第二方面，本技术公开了一种多源异构数据仓库之间的通信装置，应用于第一异构数据仓库，包括：

41、请求接收模块，用于接收包含数据源参数、第二异构数据仓库的类型及数据仓库集群标识的数据传输请求；

42、数据获取模块，用于根据所述数据仓库集群标识查询预设集群路由表，以确定与所述数据仓库集群标识对应的元数据客户端、执行引擎、分布式文件系统客户端以及资源队列；

43、查询结果获取模块，用于基于所述元数据客户端将所述数据源参数对应的查询语句改写为创建临时表的结构化查询语言语句，通过所述执行引擎基于所述资源队列限定的计算资源执行所述结构化查询语言语句，以得到查询结果；

44、当前临时表写入模块，用于通过所述执行引擎将所述查询结果物化为当前临时表，通过所述分布式文件系统客户端将当前临时表写入分布式文件系统上临时路径下的文件；

45、传输任务查找模块，用于在当前临时表写入完成后，根据所述第二异构数据仓库的类型从传输任务注册表中查找与所述第二异构数据仓库的类型对应的传输任务；

46、数据写入模块，用于在传输阶段，通过所述分布式文件系统客户端基于所述临时路径流式读取当前临时表的数据，并基于所述传输任务将读取到的数据写入第二异构数据仓库。

47、第三方面，本技术公开了一种电子设备，包括：

48、存储器，用于存储计算机程序；

49、处理器，用于执行计算机程序以实现如前述多源异构数据仓库之间的通信方法的步骤。

50、第四方面，一种计算机可读存储介质，计算机可读存储介质上存储有计算机程序，计算机程序被处理器执行时实现如前述多源异构数据仓库之间的通信方法的步骤。

51、本技术第一异构数据仓库首先接收包含数据源参数、第二异构数据仓库的类型及数据仓库集群标识的数据传输请求；根据所述数据仓库集群标识查询预设集群路由表，以确定与所述数据仓库集群标识对应的元数据客户端、执行引擎、分布式文件系统客户端以及资源队列；基于所述元数据客户端将所述数据源参数对应的查询语句改写为创建临时表的结构化查询语言语句，通过所述执行引擎基于所述资源队列限定的计算资源执行所述结构化查询语言语句，以得到查询结果；通过所述执行引擎将所述查询结果物化为当前临时表，通过所述分布式文件系统客户端将当前临时表写入分布式文件系统上临时路径下的文件；在当前临时表写入完成后，根据所述第二异构数据仓库的类型从传输任务注册表中查找与所述第二异构数据仓库的类型对应的传输任务；在传输阶段，通过所述分布式文件系统客户端基于所述临时路径流式读取当前临时表的数据，并基于所述传输任务将读取到的数据写入第二异构数据仓库。可见，本技术通过预设集群路由表统一管理不同数据仓库集群对应的元数据客户端、执行引擎、分布式文件系统客户端及资源队列，实现了对异构数据仓库集群的自动识别与底层差异屏蔽；通过将查询语句改写为创建临时表的结构化查询语言语句并借助执行引擎完成数据物化，再经由分布式文件系统客户端将临时表写入统一临时路径，构建了标准化的中间数据层，使得数据抽取与数据下发相互解耦；通过根据第二异构数据仓库的类型从传输任务注册表中查找对应的传输任务并执行数据写入，实现了对不同类型异构数据仓库的统一接入与标准化下发，有效降低了多集群多目标场景下的适配复杂度，提升了数据传输系统的可扩展性与执行效率。