[發明專利]一種并行分布式大數據架構構建方法及系統

有效

• 申請號: 202110440139.8
• 申請日: 2021-04-21
• 公開（公告）號: CN113190528B
• 公開（公告）日: 2022-12-06
• 發明（設計）人: 周生輝;劉園園;王勇;萬修全;魏志強
• 申請（專利權）人: 中國海洋大學;青島海洋科學與技術國家實驗室發展中心
• 主分類號: G06F16/21
• 分類號: G06F16/21;G06F16/22;G06F16/2455;G06F16/27
• 代理公司: 北京艾格律詩專利代理有限公司 11924
• 代理人: 王子溟

1.一種并行分布式大數據架構構建方法，其特征在于，包括生產者端的搭建及數據連接器端的搭建，所述生產者端的搭建方法包括：

在即將運行的MPI進程中加載指定的序列化策略；

通過kafka生產者MPI進程組的各個進程完成各kafka生產者的生產屬性配置；

通過所述kafka生產者MPI進程組的各個進程進行序列化策略的類型轉化，對預備發送數據進行avro序列化；

基于緩存的序列化策略元信息對avro序列化后的數據進行內存填充，并形成kafka消息，發送至kafka消息隊列中；

所述數據連接器端的搭建方法包括：

通過spark-sql-kafka集成模塊以流方式從kafka數據源讀取Dataframe，并完成預置的組態配置，其中，所述組態配置包括：自如下三個配置項中的一個作為kafka數據源的目標topic，所述配置項包括assign*、subscribe*、subscribePattern*；

從schema registry端加載指定的kafka數據源topic鍵、值avro序列化策略，并執行反序列化；

使用spark微批處理形式，將反序列化后Kafak數據流批量寫入GreenPlum目標數據庫或數據表中；所述在即將運行的MPI進程中加載指定的序列化策略包括：

創建kafka生產者MPI進程組和通信域；

選取MPI進程組中的一個MPI進程作為主進程，構建所述主進程與Schema Registry端的通信，加載指定的序列化策略；

將所述主進程中的序列化策略廣播至Kafka生產者MPI進程組中的其它所有即將運行的MPI進程；加載指定的序列化策略包括：

調用libserdes庫中的serdes_schema_get函數從指定位置進行加載，若指定位置不存在所述序列化策略，則調用libserdes庫中的serdes_schema_add函數將序列化策略定義添加至指定位置，所述指定位置為本地緩存或所述schema registry端；將所述主進程中的序列化策略廣播包括：

調用MPI_BCAST函數將序列化策略進行廣播；完成各kafka生產者的生產屬性配置，包括：

完成服務器列表的配置；

完成消息生產批處理最小本地隊列大小的配置；以及

完成消息生產本地隊列最長等待時間的配置。

2.如權利要求1所述的并行分布式大數據架構構建方法，其特征在于，進行序列化策略的類型轉化包括：

調用avro-c庫中的相關函數將本地緩存的序列化策略轉化為avro數據序列化系統相關的avro_schema_t結構體類型。

3.如權利要求1所述的并行分布式大數據架構構建方法，其特征在于，所述執行反序列化后進一步包括：

執行所需的動作、轉換、連接、分組、聚合的結構化數據操作。

4.如權利要求1所述的并行分布式大數據架構構建方法，其特征在于，將反序列化后Kafak數據流批量寫入GreenPlum目標數據庫或數據表中之前包括配置目標GreenPlum數據源，所述配置目標GreenPlum數據源包括：

調用GreenPlum-spark-connector集成模塊，使用GreenPlum數據源，完成預置的數據庫組態配置，所述數據庫組態配置包括數據庫JDBC URL、數據庫目標schema、數據表、JDBC驅動、數據庫用戶名、數據庫用戶密碼。

5.一種并行分布式大數據架構構建系統，其特征在于，包括生產者端的搭建模塊及數據連接器端的搭建模塊，所述生產者端的搭建模塊包括：

序列化策略加載單元，用于在即將運行的MPI進程中加載指定的序列化策略；

生產屬性配置單元，用于通過kafka生產者MPI進程組的各個進程完成各kafka生產者的生產屬性配置；

avro序列化單元，用于通過所述kafka生產者MPI進程組的各個進程進行序列化策略的類型轉化，對預備發送數據進行avro序列化；

kafka消息生產單元，用于基于緩存的序列化策略元信息對avro序列化后的數據進行內存填充，并形成kafka消息，發送至kafka消息隊列中；

所述數據連接器端的搭建模塊包括：

組態配置單元，用于通過spark-sql-kafka集成模塊以流方式從kafka數據源讀取Dataframe，并完成預置的組態配置，其中，所述組態配置包括：自如下三個配置項中的一個作為kafka數據源的目標topic，所述配置項包括assign*、subscribe*、subscribePattern*；

反序列化單元，用于從schema registry端加載指定的kafka數據源topic鍵、值avro序列化策略，并執行反序列化；

微批處理單元，用于使用spark微批處理形式，將反序列化后Kafak數據流批量寫入GreenPlum目標數據庫或數據表中；所述序列化策略加載單元包括：

MPI進程組及通信域假設子單元，用于創建kafka生產者MPI進程組和通信域；

指定的序列化策略加載子單元，用于選取MPI進程組中的一個MPI進程作為主進程，構建所述主進程與Schema Registry端的通信，加載指定的序列化策略；

廣播子單元，用于將所述主進程中的序列化策略廣播至Kafka生產者MPI進程組中的其它所有即將運行的MPI進程；其中，

加載指定的序列化策略包括：

調用libserdes庫中的serdes_schema_get函數從指定位置進行加載，若指定位置不存在所述序列化策略，則調用libserdes庫中的serdes_schema_add函數將序列化策略定義添加至指定位置，所述指定位置為本地緩存或所述schema registry端；將所述主進程中的序列化策略廣播包括：

調用MPI_BCAST函數將序列化策略進行廣播；完成各kafka生產者的生產屬性配置，包括：

完成服務器列表的配置；

完成消息生產批處理最小本地隊列大小的配置；以及

完成消息生產本地隊列最長等待時間的配置。