一種高可靠分布式數據流實時統計方法及系統與流程

2023-10-18 04:21:39 2

本發明涉及一種高可靠分布式數據流實時統計方法及系統，屬於大數據
技術領域：
。
背景技術：
：提到數據流統計處理，目前業界的解決方案絕大多數都是基分布式內存計算，這是因為分布式系統的並發性可以很好的應對大規模的數據流，使用內存計算而不是傳統的本地或者分布式文件系統，是因為系統需要儘快的處理流入系統的數據，因為在數據流計算場景下，數據是高頻流動的，具有時效性，當一包數據流過去就無法再次拿到這包數據。在這樣的分布式內存計算架構下，存在著諸多需要解決的問題，從一致性、可用性、分區容忍三選二的折中，到可靠性、擴展性和靈活性等等特性。現有的產品例如s4、storm、spark都已經一定程度上得到了業界的認可及應用，但是在更加苛刻的場景下，他們的表現並不理想。s4的不足主要有兩個不足，首先是可靠性，s4僅能保證at-most-once語義，當processingnode宕機後，任務可轉移，但是內存中的數據會全部丟失。另外s4不可動態擴展節點，這對於一個分布式系統來說是不可接受的。storm存在的問題有兩點：一是弱中心化結構帶來的單點故障，在nimbus節點或者ui節點發生宕機後，統計任務的執行會出現問題；二是語義保證，即可靠性，儘管提供了trident機制保證了exactly-once語義，但是該機制會嚴重影響處理性能。sparkstreaming存在的問題同樣有兩點：一是處理延遲較高，秒級的延遲無法滿足某些對實時性要求較高的應用場景；二是在語義保證上，需要藉助特定的數據源保證不丟失數據。綜上，可靠性是
背景技術：
存在的核心問題，由可靠性帶來的性能問題、兼容性問題同樣值得關注。技術實現要素：本發明的目的在於針對高精確性需求的數據流實時統計場景下，提供一個高可靠、並最大程度上保證性能(吞吐量)的數據流實時統計系統，使得在發生宕機、故障時，可以保證數據可恢復、任務可轉移。本發明的技術方案為：一種高可靠分布式數據流實時統計方法，其步驟為：1)根據當前的統計任務生成分布式集群mapreducer的map任務和reducer任務；2)針對每個計算任務為mapper集群中的每一mapper節點初始化一mapper序號、為reducer集群中的每一reducer節點初始化一reducer序號；其中，mapper序號初始化為mapper集群已經在分布式緩存中生成的中間數據集個數；reducer序號初始化為reducer集群下一輪計算即將從分布式緩存中取得的中間數據集序號；3)mapper集群中的mapper節點根據當前的mapper序號從消息中間件拉取消息並處理；其中，當一個mapper節點處理完一個時間粒度的輸入數據產生初步計算結果後，將mapper序號自增1並把初步計算結果以自增後的mapper序號存儲在分布式緩存系統redis中；4)reducer集群從分布式緩存系統redis中順序讀取所述初步計算結果並處理，得到當前統計任務的最終統計結果；其中，每個reducer節點處理任務時，從分布式緩存系統redis中得到reducer序列號，然後從分布式緩存系統redis中讀取該reducer序號的數據；如果reducer節點需要歸併多個reducer序號的數據集，則該reducer節點保存增加後的reducer序號，直到該多個reducer序號的數據集處理完成後，將該reducer節點的reducer序號更新到分布式緩存系統redis中。進一步的，reducer集群首先選取一節點作為master節點，然後該master節點使用一致性哈希算法計算各個節點處理的任務編號，並根據計算結果向各個節點發送調度信號，告知各節點需要處理的任務；各節點監聽自己的任務狀態，在接收到調度信號時啟動對應的計算任務或停止不再處理的計算任務。進一步的，選取master節點的方法為：每個reducer節點上線時在指定目錄註冊命名相同的臨時順序節點；將最先上線註冊的節點作為master節點。進一步的，該master節點使用一致性哈希算法計算各個節點處理的任務編號的方法為：該master節點首先將各reducer節點以節點ip為key註冊在哈希環上；然後對於每一個任務，將該任務的id加上混淆字符串註冊在該哈希環上；然後在該哈希環上尋找遇到的第一個節點並將該任務分配給該節點。進一步的，每個數據流統計任務有兩種狀態：等待狀態和運行狀態，以及兩種信號：就緒信號和限制信號；通過就緒信號和限制信號控制當前統計任務在等待狀態和運行狀態之間轉換。進一步的，步驟3)中，利用分布式緩存系統redis提供的自增原子操作將mapper序號自增1。進一步的，步驟4)中，在reducer集群中，每個reducer任務的一個並行度僅對應一個reducer節點。一種高可靠分布式數據流實時統計系統，其特徵在於，包括任務資料庫、消息中間件、map集群、reducer集群、分布式緩存系統redis和分布式協調系統；其中，針對每個計算任務為mapper集群中的每一mapper節點初始化一mapper序號、為reducer集群中的每一reducer節點初始化一reducer序號；其中，mapper序號初始化為mapper集群已經在分布式緩存中生成的中間數據集個數；reducer序號初始化為reducer集群下一輪計算即將從分布式緩存中取得的中間數據集序號；mapper集群，各mapper節點根據當前的mapper序號從消息中間件拉取消息並處理；當一個mapper節點處理完一個時間粒度的輸入數據產生初步計算結果後，將mapper序號自增1並把初步計算結果以自增後的mapper序號存儲在分布式緩存系統redis中；reducer集群，用於從分布式緩存系統redis中順序讀取所述初步計算結果並處理，得到當前統計任務的最終統計結果；其中，每個reducer節點處理任務時，從分布式緩存系統redis中得到reducer序列號，然後從分布式緩存系統redis中讀取該reducer序號的數據；如果reducer節點需要歸併多個reducer序號的數據集，則該reducer節點保存增加後的reducer序號，直到該多個reducer序號的數據集處理完成後，將該reducer節點的reducer序號更新到分布式緩存系統redis中；任務資料庫，用於存儲分布式集群mapreducer的map任務和reducer任務；分布式消息中間件，用於提供消息服務，以及統計任務資料庫儲存用戶的統計需求；分布式協調系統，用於提供對分布式集群運行時的狀態管理；分布式緩存系統redis，用於緩存分布式中間計算結果。本發明包括三項核心技術：1.基於mapreduce編程模型的分布式數據流計算模型2.帶序號的並發數據傳遞機制3.基於狀態與信號的分布式任務管理調度機制如圖1所示，其中技術1通過保證分布式系統的可擴展性，解決了吞吐量的問題。技術二和技術三分別解決了數據可靠性和任務可用性的問題，從而保證了可靠的語義。1基於mr的分布式數據流計算模型本發明的系統架構圖如圖2所示，粗箭頭表示數據流向，細箭頭表示組件之間的交互，mapper集群、reducer集群是mars的核心組件。mapper集群從消息中間件拉取消息並處理，將中間結果順序緩存在分布式緩存中，reducer集群從分布式緩存中順序讀取分布式緩存中的中間數據，處理後將最終的統計結果再發送回消息隊列。mars將mapreducer模型的思想擴展到了集群概念上，每一個map或者reducer計算單元都是分布式集群中的一個節點，分別稱作mapper或reducer。所有的mapper節點組成mapper集群，所有的reducer節點組成reducer集群。mars依賴的外部組件有四個：資料庫、消息中間件、分布式緩存系統redis和分布式協調服務zookeeper。其中分布式消息中間件提供消息服務，mars的輸入和輸出都通過分布式消息中間件完成，統計任務資料庫儲存用戶的統計需求，分布式協調系統提供對分布式集群運行時的狀態管理等等，分布式緩存負責異步解耦和緩存中間計算結果。在本發明中，每個數據流統計任務都分為map和reduce兩個階段，分別由mapper集群和reducer集群完成。以一個分組統計任務為例，如圖3。需要特別指出的是，這裡的map和reduce與傳統的mapreduce模型不同。傳統的mapreduce模型用於離線批處理，數據批量輸入系統，當mapper處理完全部數據後才會開始reduce階段。本發明中對mapreduce的思想進行了延伸，應用到了數據流實時處理領域。在處理流數據時，數據經過mapper處理結束後直接進入reduce階段，不必等待所有數據處理完成，保證了流數據處理的實時性。同時，本發明中每個mapper和reducer都是一個物理節點，更粗的粒度有利於節約故障恢復時的開銷。2帶序號的並發數據傳遞技術在本發明中，針對每個計算任務為mapper集群和reducer集群初始化一個序號：mapper的序號表示mapper已經在分布式緩存中生成的中間數據集個數；reducer的序號表示reducer下一輪計算即將從分布式緩存中取得的中間數據集序號。圖4是數據從mapper發送到分布式緩存的過程。mapper集群採用自增mapper序號的策略，也就是說，當一個mapper節點處理完一個時間粒度的輸入數據、產生初步計算結果後，利用redis提供的自增原子操作將mapper序號自增1，並把初步計算結果以自增後的序號存儲在redis中。由於自增為原子操作，所以多個節點並行處理同一task時，每個mapper節點所得到的序號一定是全局唯一的，不會將計算結果存儲為同一序號，造成數據覆蓋。mapper和reducer的序號策略示意圖如圖5。reducer集群使用了延遲更新reducer序列號的策略。在reducer集群中，每個任務的一個並行度僅對應一個reducer節點，每個reducer節點處理任務時，從分布式緩存中得到reducer序列號，上文中定義reducer序列號為本次計算即將要處理的數據集序號，因此reducer節點得到序號後則從redis中讀取該reducer序列號的數據，如果reducer需要歸併多個序號的數據集，reducer也僅在內存中保存增加後的序列號，直到所有序號處理完成後才將reducer序列號更新到redis。這一策略主要是出於容錯性考慮。綜上，這一技術使得mapper集群和reducer集群間以一種序號機制保證消息的順序和可靠性，保證了數據級別的可靠性，使得在mapper集群或reducer集群中發生節點宕機時，數據不會丟失。3基於狀態與信號的分布式任務管理調度技術在本發明中，每個數據流統計任務有兩種狀態：waiting和running，另外有兩種信號：ready和term可以使得兩種狀態發生轉換，轉換關係如圖6。reducer集群中master(領導者)選舉後，master使用一致性哈希算法計算各個節點處理的任務編號，並根據計算結果向各個節點發送調度信號(即節點對應的任務編號)，告知該節點需要處理的任務。所有節點監聽自己的任務狀態，在接收到調度信號時啟動對應的計算任務或停止不再處理的計算任務。同時，所有節點監聽集群節點在線狀態，在新節點上線、運行中節點發生故障時，重新校驗master身份，並由master重新分配任務並根據任務分配變化情況發送相應調度信號。reducer的master選舉機制依賴於zookeeper。zookeeper有一種臨時順序節點類型叫做ephemeral_sequential，臨時順序節點的特性是註冊節點時會在client指定的命名後添加一個序號，按照註冊該節點的順序，序號遞增。利用這一機制，reducer的領導者選舉機制實現如下：每個reducer節點上線時在指定目錄註冊命名相同的臨時順序節點(ephemeral_sequential_node)。reducer集群各個節點上線的時間是有順序的，使最先上線註冊的節點成為master。以圖7為例，r1、r2、r3為3個reducer節點，ip分別為192.168.1.1、192.168.1.2、192.168.1.3，它們上線時都去創建名為/master-election/lock的zookeeper節點，並將zookeeper節點賦值為自己的ip。reducer的一致性哈希分配算法敘述如下：首先，master節點將所有reducer節點以節點ip為key註冊在哈希環上；接著，對於每一個task，將task的id加上混淆字符串註冊在哈希環上；然後，在哈希環上尋找遇到的第一個節點；最後，將該task分配給這個節點。這些機制使得在集群規模發生變化時(新節點上線或工作節點宕機)，計算任務能夠得以正確的調度和轉移，保證了任務級別的可靠性。與現有技術相比，本發明的積極效果為：本發明能夠滿足高精確性需求的數據流實時統計，具有高可靠、並最大程度上保證性能(吞吐量)，使得在系統發生宕機、故障時，可以保證數據可恢復、任務可轉移。本發明在處理流數據統計任務，以及預處理、統計的混合任務方面的吞吐量比現有的流處理系統stom、sparkstreaming大。具體對比數據如圖8。附圖說明圖1為本發明技術架構框圖；圖2為本發明的系統架構圖；圖3為一個分組統計任務實例圖；圖4為數據從mapper發送到分布式緩存的過程圖；圖5為mapper和reducer的序號策略示意圖；圖6為轉換關係圖；圖7為註冊流程示例圖；圖8為對比數據圖；(a)數據流統計任務吞吐量對比；(b)混合任務吞吐量對比；圖9為map階段處理流程圖；圖10為reduce階段處理流程圖。具體實施方式下面結合具體實例對本發明進行進一步具體描述。示例1假設數據源數據的格式包含4個欄位，id、時間戳(timestamp)、源ip(sip)、目標ip(dip)，其中一段數據流如下：idtimestampsipdip109:251.1.1.12.2.2.2109:251.1.1.13.3.3.3109:261.1.1.16.6.6.6109:263.3.3.35.5.5.5209:274.4.4.42.2.2.2209:284.4.4.43.3.3.3209:286.6.6.61.1.1.1現需要在3min的窗口上統計出現的所有id使用某ip產生了多少條信息。使用本技術發明，將上述需求配置為一個服務規則如下：數據源map粒度reduce粒度發送到數據處理規則info_mq1min3minresult_mqgroup_by_and_count:id,sipmap節點和reduce節點都會讀取到這一規則，並將規則解析成map和reduce對應的任務。其中，數據處理規則由冒號分為兩部分：「group_by_and_count」和「id,sip」。前一部分是規則名稱，意味著我們需要對數據做分組計數的操作；後一部分是操作對象，由逗號隔開，意味著我們對「id」和「sip」兩個欄位進行分組。按照規則中的粒度，map節點在每個map粒度(1min)上將id和源ip作為判斷依據，相同的累加數量，不同的新建一個計數，各個map節點將如下的統計結果發送到分布式緩存中。109:251.1.1.12109:261.1.1.11109:263.3.3.31209:274.4.4.41209:284.4.4.41209:286.6.6.61最終，reduce節點將這些結果按照reduce的粒度(最終統計粒度)聚合起來，產生如下的最終統計結果，發送到分布式消息隊列的指定話題中。109:251.1.1.13109:263.3.3.31209:274.4.4.42209:286.6.6.61map階段處理流程如圖9所示：1.首先從配置庫中獲取需要處理的需求號以及對應規則，並初始化對應的需求處理者；2.根據需求號從消息隊列中拉取數據；3.處理完成後向消息隊列返回「消費成功」信號；4.將數據分配給各個處理者處理；5.數據處理完成後，進行如下事務操作：(1)在緩存中累加並取得該需求的mapper序列號；(2)根據該序列號將細粒度統計結果寫入分布式緩存；6.至此，mapper部分數據處理流程結束。reduce階段處理流程如圖10所示：1.從分布式緩存獲得要讀取的reducer序列號；2.根據該序列號從緩存中讀取多個細粒度統計結果；3.講數據聚合成1包最終結果；4.向消息隊列發送最終結果；5.將讀取數據後的序列號累加回分布式緩存，更新序列號；6.至此，reducer部分數據處理流程結束。當前第1頁12