[發明專利]實時數據流處理方法、裝置、設備、及可讀存儲介質

在審

• 申請號: 202011024249.8
• 申請日: 2020-09-25
• 公開（公告）號: CN112416972A
• 公開（公告）日: 2021-02-26
• 發明（設計）人: 陳健;蔡雪峰
• 申請（專利權）人: 上海嗶哩嗶哩科技有限公司
• 主分類號: G06F16/2455
• 分類號: G06F16/2455;G06F16/28
• 代理公司: 北京英特普羅知識產權代理有限公司 11015
• 代理人: 程超

本申請公開了一種實時數據流處理方法、裝置、設備、及可讀存儲介質，該方法包括：獲取多個數據流在預設時間窗口內產生的數據項；其中，每個數據項包括：一個索引字段；將所述多個數據流中的包含同一索引字段的數據項進行合并，并將合并后的合并數據項存儲到緩存中；當所述合并數據項在所述緩存中的存儲時長達到預設閾值時，將所述合并數據項存儲到目標數據庫中；本申請能夠將多條數據流合并后存儲到目標數據庫中，從而減少目標數據庫中的冗余數據。

技術領域

本申請涉及數據處理技術領域，特別涉及一種實時數據流處理方法、裝置、設備、及可讀存儲介質。

背景技術

在現有的流式技術中，多條數據流中的數據項會直接存儲到MySQL數據庫中，再基于MySQL數據庫進行后續的數據處理；通常多條數據流之間的數據項存在一定的關聯關系，但由于單一的數據流無法獲取其他數據流中的數據信息，因此在將數據流中的數據項存入到MySQL數據庫之前，只會通過算子對同一數據流中的數據項進行處理，不會對多個數據流之間的數據項進行處理，因此，會在MySQL數據庫中存儲大量冗余數據，降低了MySQL數據庫的存儲效率。

發明內容

本申請的目的在于提供一種實時數據流處理方法、裝置、設備、及可讀存儲介質，能夠將多條數據流合并后存儲到目標數據庫中，從而減少目標數據庫中的冗余數據。

根據本申請的一個方面，提供了一種實時數據流處理方法，所述方法包括：

獲取多個數據流在預設時間窗口內產生的數據項；其中，每個數據項包括：一個索引字段；

將所述多個數據流中的包含同一索引字段的數據項進行合并，并將合并后的合并數據項存儲到緩存中；

當所述合并數據項在所述緩存中的存儲時長達到預設閾值時，將所述合并數據項存儲到目標數據庫中。

可選的，在所述將合并后的合并數據項存儲到緩存中之后，且所述合并數據項在所述緩存中的存儲時長未達到所述預設閾值時，所述方法還包括：

當任一數據流中有撤回流操作時，獲取所述數據流中的更新數據項；

根據所述更新數據項中的索引字段，從所述緩存中查找到包含所述索引字段的合并數據項，并根據所述更新數據項對所述合并數據項進行更新。

可選的，所述方法還包括：

獲取預設的合并配置文件；其中，所述合并配置文件包括：多個目標索引字段、以及每個目標索引字段所對應的合并規則。

可選的，所述將所述多個數據流中的包含同一索引字段的數據項進行合并，并將合并后的合并數據項存儲到緩存中，包括：

針對所述合并配置文件中的一個目標索引字段，判斷在所述多個數據流中是否存在包含所述目標索引字段的數據項；

若是，則將所有包含所述目標索引字段的數據項發送至與所述目標索引字段對應的合并節點上；

通過所述合并節點，按照所述合并配置文件中的與所述目標索引字段對應的合并規則，將所有包含所述目標索引字段的數據項合并為合并數據項；

將所述合并數據項、以及所有包含所述目標索引字段的數據項存儲到所述合并節點上的緩存中。

可選的，所述根據所述更新數據項中的索引字段，從所述緩存中查找到包含所述索引字段的合并數據項，并根據所述更新數據項對所述合并數據項進行更新，包括：

將所述更新數據項存儲到與所述索引字段對應的合并節點上的緩存中；

確定出所述更新數據項的數據流類型；

從所述緩存中刪除包含所述索引字段、且與所述數據流類型一致的數據項，以及從所述緩存中刪除包含所述索引字段的合并數據項；