[發明專利]一種流聚合方法、裝置及電子設備

有效

• 申請號: 201910073375.3
• 申請日: 2019-01-25
• 公開（公告）號: CN111488222B
• 公開（公告）日: 2023-04-07
• 發明（設計）人: 郭峰
• 申請（專利權）人: 杭州海康威視數字技術股份有限公司
• 主分類號: G06F9/54
• 分類號: G06F9/54
• 代理公司: 北京博思佳知識產權代理有限公司 11415
• 代理人: 林祥

本申請提供一種流聚合方法、裝置及電子設備，該方法包括：響應于接收到的第一流聚合指令，獲取第一流聚合指令指示的第一消息隊列的第一消息和第二消息隊列的第二消息，并依時序分別對第一消息和第二消息進行流聚合處理；檢查對第一消息和第二消息的處理時差是否達到預設的時差閾值；其中，處理時差為處理中的第一消息攜帶的事件時間與第二消息攜帶的事件時間的差值；若是，向處理速度慢的消息隊列新增用于流聚合處理的線程，或者，減少用于對處理速度快的消息隊列進行流聚合處理的線程；對所述第一消息的處理結果和所述第二消息的處理結果進行關聯操作。在本申請中，有效地實現了對兩個消息隊列的流聚合處理結果的關聯操作，更好地呈現處理結果。

技術領域

本申請涉及大數據分析領域，特別涉及一種流聚合方法、裝置及電子設備。

背景技術

Kafka集群作為消息處理中心，負責消息的存儲、確認和重試等任務，Kafka集群上包含多個消息隊列(Message?Queue)，其中，消息隊列中的消息由消息生產者產生并發送至消息處理中心。消息消費者從消息處理中心獲取消息，然后進行相應的處理。通常而言，作為消息消費者的設備可以對消息隊列的數據流進行流聚合(Stream?Aggregate)處理，即對一個或多個消息隊列中的消息基于時間區間分組后，按照指定的聚合函數計算各時間區間內消息，常見的聚合函數包括SUM、COUNT、AVG、MIN、MAX等。

當消息消費者對至少兩個消息隊列中的消息進行流聚合處理時，為更好地呈現處理結果，通常可對兩個消息隊列的處理結果進行關聯操作(Join?Operations)。由于對不同消息隊列的處理速度可能不同，或者，從消息處理中心獲取不同消息隊列的消息的速度可能不同，因此，在流聚合處理的過程中，需將不同消息隊列的各個時間區間內的消息進行對齊，從而能同時(或者在很短的時間內)獲得同一時間區間不同消息隊列的消息的流聚合處理的處理結果，進而對不同消息隊列的處理結果進行關聯操作。

在相關技術中，作為消息消費者的設備可以基于預設的容忍延遲時間來延遲輸出處理速度較快的數據流，進而在輸出數據流之前，可以將處理速度較快數據流與處理速度較慢的數據流中處于同一時間區間的消息的流聚合處理結果進行關聯操作。

然而，容忍延遲時間往往是運營人員的經驗值，并不能滿足實際需求。若容忍延遲時間過長，則數據流中延遲輸出的消息的數據量會超過設備的緩存空間；若容忍延遲時間過短，則可能存在兩個數據流的處理時差大于容忍延時時間，這就導致無法實現流對齊，也就無法實現同時對兩個數據流進行流聚合處理。

發明內容

有鑒于此，本申請提供一種流聚合的方法、裝置及電子設備，用以實現同時對兩個數據流進行流聚合處理。

具體地，本申請是通過如下技術方案實現的：

一種流聚合方法，包括：

響應于接收到的第一流聚合指令，獲取所述第一流聚合指令指示的第一消息隊列的第一消息和第二消息隊列的第二消息，并依時序分別對所述第一消息和所述第二消息進行流聚合處理；

檢查對所述第一消息和所述第二消息的處理時差是否達到預設的時差閾值；其中，所述處理時差為處理中的第一消息攜帶的事件時間與第二消息攜帶的事件時間的差值；

若是，向處理速度慢的消息隊列新增用于流聚合處理的線程，或者，減少用于對處理速度快的消息隊列進行流聚合處理的線程；

對所述第一消息的處理結果和所述第二消息的處理結果進行關聯操作。

在所述流聚合方法中，所述方法還包括：

響應于接收到的第二流聚合指令，確定所述第二流聚合指令中至少兩個流聚合時間段的先后順序；

獲取所述第二流聚合指令指示的第三消息隊列的第三消息，依時序對所述第三消息進行流聚合處理；