王中華,柴小麗

(中國電子科技集團公司第三十二研究所,上海 201808)

隨著信息科學(xué)和云計算技術(shù)的飛速發(fā)展,智能設(shè)備的持續(xù)普及,存儲設(shè)備性能的提升和網(wǎng)絡(luò)帶寬的增長為大數(shù)據(jù)的存儲和流通提供了物質(zhì)基礎(chǔ),云計算技術(shù)通過將分散的數(shù)據(jù)集中在數(shù)據(jù)中心,從而可以更為集中有效地處理和分析大數(shù)據(jù)信息.云計算技術(shù)為海量數(shù)據(jù)存儲和分散的用戶訪問提供了必要的空間和途徑[1].大數(shù)據(jù)主要有4 種計算模式,分別是批量計算,流式計算,圖計算和交互計算.其中,適用于大數(shù)據(jù)分析的計算模式主要是批量計算和流式計算這兩種,并且由于批量計算和流式計算針對的數(shù)據(jù)流類型不同,所適用的大數(shù)據(jù)應(yīng)用場景也不一樣[2].

批量計算會將先將數(shù)據(jù)信息統(tǒng)一收集起來,然后把大量的數(shù)據(jù)信息存儲到本地或云端地數(shù)據(jù)庫中,最后對數(shù)據(jù)進(jìn)行批量的處理.由此可見,批處理適用的數(shù)據(jù)一般是靜態(tài)數(shù)據(jù),即保存在本地或云端數(shù)據(jù)庫中的信息,任務(wù)可一次性完成.因此,批量計算一般應(yīng)用在實時性要求不高,離線計算的場景下,進(jìn)行數(shù)據(jù)分析或?qū)崿F(xiàn)離線報表等[3–5].

數(shù)據(jù)流是一組有序的,有起點和終點的字節(jié)的數(shù)據(jù)序列,一般包含輸入流和輸出流,在實時通信領(lǐng)域中,數(shù)據(jù)的價值與時間成反比,即處理數(shù)據(jù)的時間越短,數(shù)據(jù)的價值越大.因此,必須對實時的數(shù)據(jù)信息給出毫秒級的響應(yīng).流式計算就是應(yīng)用在實時場景下,或?qū)r效性要求比較高的場景,如實時推薦,業(yè)務(wù)監(jiān)控[6–9].

Hadoop是一個由Apache 基金會開發(fā)的分布式系統(tǒng)基礎(chǔ)架構(gòu),實現(xiàn)了一個分布式文件系統(tǒng)HDFS,提供高吞吐量來訪問應(yīng)用程序的數(shù)據(jù),適合有著超大數(shù)據(jù)集的應(yīng)用程序,HDFS為海量數(shù)據(jù)提供存儲,MapReduce為海量數(shù)據(jù)提供計算[10].Apache 開發(fā)Flink 開源流處理框架,核心是用Java和Scala 編寫的分布式數(shù)據(jù)流引擎,Flink 通過支持?jǐn)?shù)據(jù)并行和流水線方式,可執(zhí)行任意流數(shù)據(jù)程序[11],Flink的流水線運行方式支持系統(tǒng)執(zhí)行批處理和流處理程序,同時支持迭代算法的執(zhí)行.Twitter 開發(fā)了Storm 框架提供分布式的,高容錯的實時計算系統(tǒng),支持多語言:Java,Python,Ruby 等[12],可實現(xiàn)亞秒級的低延遲.Storm 提供較高的可靠性,所有信息都可保證至少處理一次,確保不會丟失信息.

由于在項目研發(fā)過程中,需要8 臺機器對視頻流執(zhí)行解碼任務(wù),另外8 臺執(zhí)行目標(biāo)識別的推理任務(wù),并且由于兩組機器都缺乏硬件條件完成其他任務(wù).傳統(tǒng)的平均分配和單機任務(wù)指定策略不能滿足項目需求,本文通過設(shè)計新調(diào)度算法來實現(xiàn)對兩組機器的任務(wù)分配.

1 Storm 模型架構(gòu)

如圖1所示,Storm 集群采用主從式架構(gòu),集群中的節(jié)點主要分為以下4 類[13]:

主節(jié)點(Master node):通過運行Storm nimbus 命令啟動Storm的主節(jié)點,nimbus是Storm 系統(tǒng)的主控節(jié)點,主要用于向集群中提交作業(yè),通過讀取ZooKeeper的工作信息分配集群的任務(wù),以及監(jiān)控整個集群狀態(tài)(有進(jìn)程級的也有線程級別的).

工作節(jié)點(Worker node):通過運行Storm supervisor命令啟動Storm的工作節(jié)點.通過設(shè)定的端口與Zoo-Keeper 進(jìn)行信息交互,讀取主控節(jié)點分配的任務(wù)信息,下載作業(yè)副本,管理屬于自己的Worker 進(jìn)程,如啟動,暫?；虺蜂N任務(wù)的工作進(jìn)程及其線程.一個工作進(jìn)程中可運行多個線程,每個線程中又可運行多個任務(wù)(task).

控制臺節(jié)點(Web console node):通過運行Storm ui 命令啟動用戶界面服務(wù)節(jié)點,默認(rèn)的服務(wù)端口號為8080,可在storm.yaml 中設(shè)定ui.port 進(jìn)行修改.可以在ui 界面上查看已提交的作業(yè)狀態(tài),包括集群的整體狀態(tài),已使用的節(jié)點數(shù),每個節(jié)點的運行情況和作業(yè)的執(zhí)行狀態(tài),支持手動停止正在執(zhí)行的作業(yè).

協(xié)調(diào)節(jié)點:通過運行ZooKeeper server start 命令啟動ZooKeeper 進(jìn)程的節(jié)點,實現(xiàn)numbus和supervisor之間的協(xié)調(diào)管理,包含分布式狀態(tài)維護和分布式配置管理等.

圖1 Storm 模型架構(gòu)

2 ZooKeeper

Storm 使用ZooKeeper 來保證集群的一致性.Storm的所有的狀態(tài)信息都是保存在ZooKeeper 里面,nimbus通過在ZooKeeper 上面寫狀態(tài)信息來分配任務(wù)[14].

Supervisor,task 通過從ZooKeeper 中讀狀態(tài)來領(lǐng)取任務(wù),同時supervisor,task 也會定期發(fā)送心跳信息到ZooKeeper,使得nimbus 可以監(jiān)控整個Storm 集群的狀態(tài),從而可以重啟一些停止的task.

在每臺機器上設(shè)置myid 文件來分配機器在集群中的id,如圖2所示,51 表示式第幾臺服務(wù)器,10.0.0.1表示服務(wù)器的IP 地址,2888:3888 表示服務(wù)器中與集群中l(wèi)eader 交換信息的端口.

圖2 拓?fù)浣Y(jié)構(gòu)

3 Topology 結(jié)構(gòu)

Topology是Storm 中運行的一個實時應(yīng)用程序,因為各個組件間的消息流動形成邏輯上的一個拓?fù)浣Y(jié)構(gòu)[15],如圖2所示.主要由以下幾部分組成:

Spout:在一個topology 中產(chǎn)生源數(shù)據(jù)流的組件.通常情況下spout 會從外部數(shù)據(jù)源中讀取數(shù)據(jù),然后轉(zhuǎn)換為topology 內(nèi)部的源數(shù)據(jù).Spout是一個主動的角色,其接口中有個nextTuple()函數(shù),Storm 框架會不停地調(diào)用此函數(shù),源源不斷地發(fā)送數(shù)據(jù).Spout 另一個重要的方法時ack和fail,Storm 監(jiān)控到tuple 從spout 發(fā)送到toplogy 成功完成或失敗時調(diào)用ack和fail,保證數(shù)據(jù)的可靠性.

Bolt:在一個topology 中接受數(shù)據(jù)然后執(zhí)行處理的組件.Bolt 可以執(zhí)行過濾,函數(shù)操作,合并,寫數(shù)據(jù)庫等任何操作.Bolt是一個被動的角色,其接口中有個execute(Tuple input)函數(shù),在接受到消息后會調(diào)用此函數(shù),用戶可以在其中執(zhí)行自己想要的操作[16].

Tuple:Storm spout,bolt 組件消息傳遞的基本單元(數(shù)據(jù)模型),Tuple是包含名稱的列表,Storm 支持所有原生類型,字節(jié)數(shù)組為Tuple 字段傳遞,如果要傳遞自定義對象,需要實現(xiàn)接口serializer[17].

Stream:源源不斷傳遞的Tuple 就組成了stream.

4 任務(wù)調(diào)度

4.1 傳統(tǒng)任務(wù)調(diào)度

Storm 集群默認(rèn)的調(diào)度器是EventScheduler[18–21],采用輪詢策略來搜索集群中所有拓?fù)浣Y(jié)構(gòu)的工作節(jié)點,將資源較為均勻的分配給任務(wù)進(jìn)程.具體分配流程如下:

先由nimbus 來計算拓?fù)涞墓ぷ髁?及計算多少個task,task的數(shù)目是指spout和bolt的并發(fā)度的分別的和nimbus 會把計算好的工作分配給supervisor 去做,工作分配的單位是task,即把計算好的一堆task 分配給supervisor 去做,即將task-id 映射到supervisor-id+port 上去,具體分配算法如算法1.

算法1.傳統(tǒng)任務(wù)調(diào)度算法1)從ZooKeeper 上獲得已有的assignment(新提交的topology為空).2)查找所有可用的slot,slot 就是可用的worker,在所有supervisor上配置的多個worker的端口.3)將任務(wù)均勻地分配給可用的worker,supervisor 會根據(jù)nimbus 分配給他的任務(wù)信息來讓自己的worker 做具體的工作,worker 會到ZooKeeper 上去查找給他分配了哪些task,并且根據(jù)這些task-id 來找到相應(yīng)的spout/bolt,它還需要計算出這些spout/bolt 會給哪些task 發(fā)送消息,然后建立與這些task的連接,然后在需要發(fā)消息的時候就可以給相應(yīng)的task 發(fā)消息.

4.2 多任務(wù)分組調(diào)度策略

而在當(dāng)前項目應(yīng)用過程中,需要將視頻解碼和目標(biāo)識別任務(wù)分別運行在兩組機器上,并且由于任務(wù)的硬件需求,解碼的任務(wù)不能在處理目標(biāo)識別的機器上運行.

為了實現(xiàn)多任務(wù)分組調(diào)度,實現(xiàn)了算法2.

算法2.多任務(wù)分組調(diào)度算法1)從ZooKeeper 上獲得已有的assignment(新提交的topology為空).2)在配置集群時為每臺機器設(shè)置supervisor 名稱,如下圖所示,通過supervisor.scheduler.meta 設(shè)置節(jié)點名稱.3)采用循環(huán)的方式,通過判斷meta.get(“name”)==”supervisor51”或meta.get(“name”)==”supervisor52”等得到匹配的supervisor 列表.4)通過componentToExecutors.get("decode")獲得解碼任務(wù)的線程數(shù).5)通過getAvailableSlots 函數(shù)提取上述指定supervisor的所有可用節(jié)點.6)構(gòu)建map＜WorkSlot,List＜ExecutorDetails>>將可用節(jié)點與線程情況相匹配.7)通過cluster.assign 函數(shù)將匹配情況提交給集群,集群將按照對應(yīng)關(guān)系分配線程,其余任務(wù)采用平均分配,由于已經(jīng)將解碼機器組的可用節(jié)點全部占滿,剩余的推理任務(wù)將自動均勻地分配到推理機器組.

5 實驗環(huán)境與結(jié)果

5.1 實驗環(huán)境

實驗共使用16 臺服務(wù)器,且均使用Linux 系統(tǒng),其中8 臺服務(wù)器作為視頻解碼組,搭載Arm v8 多核處理器和中科睿芯解碼卡;另外8 臺服務(wù)器作為推理組,搭載Arm v8 多核處理器和寒武紀(jì)MUL100 加速卡.

Storm 集群由1 個nimbus和16 個supervisor 節(jié)點組成(為了實現(xiàn)資源充分使用,其中一個服務(wù)器既作為nimbus 用來分發(fā)任務(wù)和監(jiān)控集群狀態(tài),也用來處理任務(wù)).

拓?fù)浣Y(jié)構(gòu):

Spout:用于讀取視頻文件作為輸入,組件命名為filename-reader,共8 個線程;

Bolt1:用于實現(xiàn)視頻解碼任務(wù),組件命名為decode,共23 個線程;

Bolt2:用于實現(xiàn)推理任務(wù)(目標(biāo)識別),組件命名為inference,共32 個線程.

5.2 實驗結(jié)果

(1)任務(wù)分配情況

由表1可知,視頻解碼組同時執(zhí)行文件讀取和視頻解碼任務(wù),推理組全部執(zhí)行推理任務(wù)(目標(biāo)識別),具體ui 結(jié)果如圖3所示.其中,在視頻解碼組中預(yù)留一個節(jié)點,用于預(yù)防解碼任務(wù)時可能出現(xiàn)的節(jié)點阻塞.

(2)作業(yè)執(zhí)行情況

由圖4知整體拓?fù)涞臄?shù)據(jù)處理情況,10 分鐘時處理了48 685 條數(shù)據(jù),3 小時處理了872 481 條數(shù)據(jù).整個拓?fù)溥\行24 小時并未發(fā)生中斷,可見調(diào)度算法的穩(wěn)定性.

表1 集群任務(wù)分配情況

圖3 多任務(wù)分組調(diào)度結(jié)果

圖4 作業(yè)執(zhí)行情況

6 結(jié)束語

本文研究了Storm 環(huán)境下,多任務(wù)在兩組機器上分別運行并存在信息交互的情況,提出了多任務(wù)分組調(diào)度策略,該機制可以將存在不同需求的兩個任務(wù)分別分配到對應(yīng)的機器群組中,以達(dá)到運行和資源分配最優(yōu)情況.實驗證明,該調(diào)度機制可實現(xiàn)視頻解碼和推理任務(wù)的分組運行,并且通過持續(xù)運行拓?fù)?4 小時,驗證了該調(diào)度機制的穩(wěn)定性.后續(xù)的工作將繼續(xù)完善該調(diào)度機制,當(dāng)存在3 個或更多任務(wù)需要指定機器群組資源運行時,能夠?qū)崿F(xiàn)多個指定任務(wù)分配到指定機器群組中,以實現(xiàn)資源的正確分配與最優(yōu)分配.