基于Storm平臺(tái)的多任務(wù)分組調(diào)度策略與實(shí)現(xiàn)①

王中華,柴小麗

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

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

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

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

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

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

1 Storm 模型架構(gòu)

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

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

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

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

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

圖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ù),同時(shí)supervisor,task 也會(huì)定期發(fā)送心跳信息到ZooKeeper,使得nimbus 可以監(jiān)控整個(gè)Storm 集群的狀態(tài),從而可以重啟一些停止的task.

在每臺(tái)機(jī)器上設(shè)置myid 文件來分配機(jī)器在集群中的id,如圖2所示,51 表示式第幾臺(tái)服務(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ùn)行的一個(gè)實(shí)時(shí)應(yīng)用程序,因?yàn)楦鱾€(gè)組件間的消息流動(dòng)形成邏輯上的一個(gè)拓?fù)浣Y(jié)構(gòu)[15],如圖2所示.主要由以下幾部分組成:

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

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

Tuple:Storm spout,bolt 組件消息傳遞的基本單元(數(shù)據(jù)模型),Tuple是包含名稱的列表,Storm 支持所有原生類型,字節(jié)數(shù)組為Tuple 字段傳遞,如果要傳遞自定義對象,需要實(shí)現(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é)點(diǎn),將資源較為均勻的分配給任務(wù)進(jìn)程.具體分配流程如下:

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

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

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

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

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

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

5 實(shí)驗(yàn)環(huán)境與結(jié)果

5.1 實(shí)驗(yàn)環(huán)境

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

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

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

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

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

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

5.2 實(shí)驗(yàn)結(jié)果

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

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

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

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

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

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

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

6 結(jié)束語

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