霍　達，宋　利

(上海交通大學 電子工程系 圖像通信與網(wǎng)絡工程研究所，上海 200240)

基于Celery的分布式視頻計算處理框架

霍達，宋利

(上海交通大學 電子工程系 圖像通信與網(wǎng)絡工程研究所，上海 200240)

摘要:為了實現(xiàn)能夠快速高效的使用計算集群解決視頻計算問題，提出一種基于Celery的分布式視頻處理框架，該框架借鑒了Hadoop的架構(gòu)設計，提出了API服務器層，JobTracker和TaskTracker的三層架構(gòu)，并對其進行了優(yōu)化使之能夠兼容計算資源與存儲資源的橫向擴展和新的計算應用的縱向擴展。詳細描述了框架的架構(gòu)設計和優(yōu)化設計。實驗數(shù)據(jù)證明分布式計算框架能夠高效地利用多節(jié)點實現(xiàn)視頻計算。

關(guān)鍵詞:分布式計算；云計算；轉(zhuǎn)碼

1分布式視頻計算研究現(xiàn)狀

分布式計算由于其易于進行橫向擴展的特性，通常被用來處理海量數(shù)據(jù)。目前也有不少研究試圖采用分布式計算框架Hadoop[1]進行視頻處理，基于Hadoop的轉(zhuǎn)碼器都采用了類似的架構(gòu)：使用FFmpeg分割合并視頻，同時在Map任務中調(diào)用FFmpeg進行轉(zhuǎn)碼。為了提高基于Hadoop的視頻計算框架的效率，研究者們做了許多工作，比如文獻[2]分析了在分布式轉(zhuǎn)碼中數(shù)據(jù)本地化的重要性，據(jù)此提出了基于數(shù)據(jù)本地化的調(diào)度策略，文獻[3]提出了一種能夠負載均衡的文件存儲系統(tǒng)，使用該系統(tǒng)能夠提高分布式文件系統(tǒng)的工作效率，文獻[4]提出了一種任務槽配置算法，能夠使得I/O資源和計算資源能夠并行化利用，有效提高了轉(zhuǎn)碼系統(tǒng)的整體效率。這些策略試圖集中解決兩個問題，其一是工作節(jié)點中I/O等待和CPU計算的并行化設計，其二是提高HDFS的讀寫效率，其優(yōu)化的思路值得借鑒，但是限于Hadoop的架構(gòu)限制，很難取得進一步的優(yōu)化。同時基于Hadoop的分布式視頻處理框架對于功能性擴展限制很大，不能很好地兼容視頻計算的一些特點。這兩點使得Hadoop難以成為通用的視頻計算框架。

Celery是一個使用簡單、易于擴展并且可靠的分布式隊列，它被設計用以解決軟件設計中的任務的異步執(zhí)行，使用分布式隊列可以將軟件中各個模塊解耦?？蚣茉O計中應用分布式隊列連接各個模塊，保證了系統(tǒng)的伸縮性。

2架構(gòu)設計

2.1分層設計

分布式計算框架共分三層，分別是API服務器層，作業(yè)管理層和任務執(zhí)行層。構(gòu)建在Django上的API服務器為終端用戶提供RESTful API接口，同時調(diào)用JobTracker的API，發(fā)起、撤銷和查詢一次作業(yè)。JobTracker管理了每一次作業(yè)，并將一次作業(yè)切分成許多的任務，并分配給對應的TaskTracker，TaskTracker負責執(zhí)行一次具體的任務。三者在邏輯上行使了不同職責，每兩層之間使用Celery進行解耦，這保證了JobTracker和TaskTracker的動態(tài)伸縮性。在接下來的小節(jié)中，為了表述的方便，本文將由后端發(fā)起的一次任務稱為一次作業(yè)(Job)，將一次具體的視頻處理操作稱為一個算子(Operator)，一個或多個算子迭代操作為一次任務(Task)。

分布式視頻處理系統(tǒng)的分層設計如圖1所示。

圖1　處理框架層級關(guān)系

每層的具體職責如下：

1)API服務器

API服務器使用了RESTful的架構(gòu)風格，在JobTracker提供的作業(yè)功能之上封裝了用戶管理，對終端用戶提供API，功能涉及到了作業(yè)管理和用戶管理兩類，作業(yè)管理包括了作業(yè)注冊、進度查詢、結(jié)果查詢、歷史作業(yè)查詢和失敗原因查詢，用戶管理使系統(tǒng)支持支持用戶私有作業(yè)，用戶私有文件管理。

2)作業(yè)管理層(JobTracker)

作業(yè)管理層具體管理了一次作業(yè)，負責的任務有：(1)分析API服務器的調(diào)用字符串，將任務拆分成迭代算子序列；(2)將完整的視頻按照固定時間或者固定大小切分成n片；(3)對n個分片執(zhí)行分片上傳→處理分片→從緩存服務器下載分片的流水線；(4)定時查詢n條流水線的執(zhí)行狀態(tài)，整合n條分片執(zhí)行情況并存入Redis狀態(tài)服務器中；(5)所有流水線完成后合并分片。

3)任務執(zhí)行層(TaskTracker)

傳統(tǒng)的分布式處理計算框架如Hadoop，用戶自行編寫Map、Reduce代碼，每一個加入到Hadoop集群的計算節(jié)點都是等價的。在分布式視頻處理框架當中，為了執(zhí)行速度，算子調(diào)用第三方庫進行運算，框架完成調(diào)度，由于軟件部署以及硬件支持的原因(比如GPU)，計算節(jié)點之間不等價。為了解決這個問題，計算框架將一些算子集合劃分為一個算子族，稱為TaskTracker集合，算子族之間保持互斥，沒有一個算子同時隸屬于兩個TaskTracker集合，一個計算節(jié)點由于其部署方式，可能可以執(zhí)行一個或多個算子族，同時，一個TaskTracker集合之內(nèi)可能有多個計算節(jié)點。算子、計算節(jié)點和TaskTracker集合三者的關(guān)系如圖2、圖3所示。任務執(zhí)行的時候由框架選定算子族保證算子可以被執(zhí)行，為了保證算子序列執(zhí)行的速度，每個TaskTracker集合內(nèi)的算子將被迭代執(zhí)行(中間變量被存儲在任務處理模塊的私有存儲中，而非返回到公有數(shù)據(jù)存儲)。TaskTracker集合中的每個算子都類似于DirectShow中的一個Transform Filter，算子序列迭代結(jié)束后，任務處理模塊向作業(yè)管理層返回狀態(tài)。一次作業(yè)對應一個或多個分片，每個分片對應一個或多個不同的TaskTracker集合，一個TaskTracker集合對應了一個或多個算子。

圖2　算子與TaskTracker關(guān)系示意

圖3　計算節(jié)點與TaskTracker關(guān)系

2.2信息流設計

在分布式系統(tǒng)中，本文設計了3個流實現(xiàn)分布式系統(tǒng)中的信息流動：

1)控制流

所有的控制流都是由Celery進行調(diào)度的，控制流的特點是無返回值，并且調(diào)用者無法明確得知調(diào)用任務執(zhí)行者的詳細狀態(tài)，細節(jié)對于雙方不透明。使用Celery調(diào)度的優(yōu)點在于，利用其訂閱特性，框架可以動態(tài)地增加JobTracker和TaskTracker以保證動態(tài)伸縮性。

2)狀態(tài)流

在分布式任務處理框架中，控制流是單向的，計算框架需要使用狀態(tài)流描述一個任務的詳細狀態(tài)。狀態(tài)流與控制流反向，從TaskTracker到JobTracker，最后回到API服務器。實現(xiàn)機制如下，TaskTracker統(tǒng)計算子序列的執(zhí)行進度，每一個算子的執(zhí)行時間，如果某個算子執(zhí)行失敗，查詢失敗原因。JobTracker查詢每一個分片流水線當前所在的TaskTracker的狀態(tài)，匯總，并以Task的UUID為Key存在Redis中。API服務器可以根據(jù)Task的UUID查詢Job Tracker的狀態(tài)信息。

3)數(shù)據(jù)流

在分布式轉(zhuǎn)碼中，區(qū)別于控制流和狀態(tài)流，本文將視頻文件統(tǒng)稱為數(shù)據(jù)，一次作業(yè)中，時間的消耗主要來自于TaskTracker執(zhí)行算子的CPU時間和數(shù)據(jù)流的IO時間。對于分布式系統(tǒng)而言，數(shù)據(jù)流的I/O設計決定了框架的執(zhí)行效率。系統(tǒng)中對于數(shù)據(jù)的操作有兩類：一類是原始文件和完成文件的管理，這些數(shù)據(jù)以靜態(tài)文件的形式儲存在API服務器的硬盤中；另一類是緩存分片文件的緩存，TaskTracker使用了迭代計算的策略，算子的臨時結(jié)果會被儲存在私有存儲當中，不同迭代組通過公有Redis交換數(shù)據(jù)。

圖4描述了一次典型的作業(yè)中，一個分片數(shù)據(jù)的時序圖。假設這個分片需要依次執(zhí)行4個算子，其中前3個算子可以被一個TaskTracker迭代執(zhí)行，這3個算子的中間結(jié)果被放在當前TaskTracker所在的物理機上，中間結(jié)果儲存在硬盤下緩存目錄中。在這個實例當中，強制組內(nèi)算子迭代的設計使TaskTracker減少了兩次對于公有存儲的讀寫。迭代的設計降低了公有臨時存儲的并發(fā)數(shù)，節(jié)省了網(wǎng)絡I/O的時間。當分布式系統(tǒng)逐漸擴展，公有存儲節(jié)點將承受非常大的負載，其I/O將成為系統(tǒng)的瓶頸，為了解決公有存儲節(jié)點的I/O負載問題，Redis的部署應該使用集群化配置。

圖4　緩存分片時序圖

2.3時序圖

圖5描述了一次典型的作業(yè)過程。這個例子當中，需要將一個視頻先去噪，再編碼。視頻需要經(jīng)過兩個算子，第一個為去噪算子，第二個為轉(zhuǎn)碼算子，執(zhí)行順序為去噪→轉(zhuǎn)碼，為了簡化時序圖，圖中暫不涉及TaskTracker內(nèi)的算子迭代。分布式系統(tǒng)有一個JobTracker實例，3個TaskTracker實例，其中一個TaskTracker可以執(zhí)行轉(zhuǎn)碼算子，兩個TaskTracker可以執(zhí)行去噪算子。

圖5　計算節(jié)點與TaskTracker關(guān)系

由API服務器發(fā)起一次作業(yè)，作業(yè)請求進入異步消息隊列等待空閑的JobTracker，等到空閑的JobTracker后，JobTracker開始執(zhí)行一次作業(yè)，首先視頻被切分為2個分片，兩個分片進入上傳→去噪TaskTracker→轉(zhuǎn)碼TaskTracker→下載的流水線。分片#1和分片#2依此進入兩個空閑的去噪TaskTracker，并執(zhí)行算子，#1和#2進入去噪模塊的時間差是分片上傳至公有數(shù)據(jù)存儲的時間，分片#1執(zhí)行去噪完畢后，進入轉(zhuǎn)碼模塊，由于轉(zhuǎn)碼TaskTracker只有一個，所以兩個分片排隊執(zhí)行。兩個分片依此轉(zhuǎn)碼結(jié)束后，回到JobTracker執(zhí)行合并操作。整個過程中，API服務器在響應上層用戶的查詢請求時會對JobTracker的狀態(tài)進行查詢，JobTracker對于TaskTracker的監(jiān)控是定時的。

3性能優(yōu)化

3.1JobTracker的流水線設計

為了解決JobTracker的I/O問題，JobTracker使用了流水線的調(diào)度方式。圖6是流水線的時序圖。

圖6　JobTracker 流水線時序圖

如果不使用流水線，上傳過程將共享帶寬，假設分片個數(shù)為n，一次上傳時間為Tupload，則一次Job中耗的時間為

旅游者離開自己生活的“第一空間”而到異地的“第二空間”進行休閑，其動機往往是放松，且周圍多是陌生人，因此對自身的要求以及對規(guī)則的遵循產(chǎn)生弱化，其結(jié)果就是對身的行為有無意識的降低要求，于是，就有了在客源地并不常見的不文明行為，如隨地扔垃圾、穿著不得體、大聲喧嘩等。

(n-1)Tupload

(1)

如果分片處理時間相仿，則下載分片時會出現(xiàn)擁堵現(xiàn)象。消耗時間有可能會更高。使用流水線解決了JobTracker內(nèi)的同步問題， JobTracker間的并行機制比較復雜，且單臺物理機的I/O上限難以突破，為了解決系統(tǒng)擴展時的I/O瓶頸問題，可以使用多API服務器實例+Nginx反向代理的方式進行負載均衡。

3.2算子的切分粒度設計

TaskTracker執(zhí)行JobTracker分配的算子序列，算子的設計有以下兩種原則，具體在實現(xiàn)中使用哪一種原則取決于應用的傾向：

1)為了提高執(zhí)行效率，分布式框架中的算子應該維持一個比較大的任務切分粒度。此時的算子為比較復雜的計算，或者是多個簡單的算子組成一個比較大的算子，迭代在算子內(nèi)部迭代執(zhí)行。

2)為了提高擴展性，此時算子切分粒度最小，每個算子只執(zhí)行一件任務，任務在TaskTracker內(nèi)迭代執(zhí)行，雖然效率比算子內(nèi)部迭代執(zhí)行低，但是算子之間的組合清晰，對于任務的擴展友好。

TaskTracker運行在計算節(jié)點上，最大化利用計算節(jié)點資源是TaskTracker設計的主要問題，TaskTracker的執(zhí)行耗時主要有，算子執(zhí)行，消耗CPU計算資源；中間結(jié)果的I/O，消耗硬盤I/O資源；分片下載與上傳，消耗網(wǎng)絡I/O資源；在實際部署時，單物理節(jié)點部署應啟動多TaskTracker實例，實例的個數(shù)以達到CPU，硬盤I/O，網(wǎng)絡I/O中的任意一個瓶頸為準，這區(qū)別于物理機性能，算子類型。

3.3失效轉(zhuǎn)移與負載均衡

4實驗結(jié)果

本節(jié)通過實驗研究分布式視頻處理框架的性能，實驗環(huán)境為HP Z820工作站3臺，CPU Intel Xeon E5-2698 v2 2.7 GHz，內(nèi)存32 Gbyte。本文挑選了一個原始分辨率為3 840×2 160，碼率140 kbit/s，長度為2 min 13 s，幀率為30 f/s(幀/秒)的視頻作為基準視頻，以計算復雜度為區(qū)分，確定了3個計算復雜度各異的任務：

1)Benchmark A： 分辨率640×360，碼率為8 000 kbit/s，使用libx264單線程進行壓縮，壓縮參數(shù)：分辨率640×360，碼率為8 000 kbit/s，直接調(diào)用libx264進行壓縮時耗時為99.89 s。

2)Benchmark B： 分辨率為1 280×720，碼率為20 000 kbit/s，使用libx264單線程進行壓縮，壓縮參數(shù)：分辨率1 280×720，碼率為20 000 kbit/s，直接調(diào)用libx264進行壓縮時耗時為380.34 s。

3)Benchmark C： 分辨率為1 920×1 080，碼率為40 000 kbit/s，使用libx264單線程進行壓縮，壓縮參數(shù)：分辨率1 920×1 080，碼率為40 000 kbit/s，直接調(diào)用libx264進行壓縮時耗時為851.72 s。

本文選定Benchmark C作為轉(zhuǎn)碼任務，測試了隨著工作節(jié)點的變化，執(zhí)行的時長，結(jié)果如圖7所示，可以看到，隨著工作節(jié)點的增加，計算耗時逐漸降低。

圖7　工作節(jié)點數(shù)改變時計算時間的變化

相比于與直接執(zhí)行算子，調(diào)用計算框架進行計算時會有額外的時間損耗，主要來自于分片的上傳下載、分片切割、合并，單個計算節(jié)點的工作效率理論上只能接近100%。下面本文使用計算的工作效率進行分析：規(guī)定當直接調(diào)用算子時，效率為1，分布式計算時，以直接執(zhí)行算子處理相同任務的時間為基準，單節(jié)點效率μ的計算公式如下

(2)

式中：nnode為節(jié)點數(shù)；Toverall為分布執(zhí)行時間；Tdirect為直接執(zhí)行算子耗時。

表1描述了不同Benchmark下工作節(jié)點數(shù)關(guān)系與效率值的關(guān)系，可以看到兩個趨勢：

1)計算越復雜，效率越高，這主要是因為影響效率值的主要因素是I/O損耗，在I/O損耗一定時，如果計算越復雜，CPU計算時間長，效率提升，這也是分布式計算框架的主要應用方向，即高CPU負載的計算任務。

2)隨著計算節(jié)點個數(shù)的增加，效率逐漸降低，這主要是因為實驗環(huán)境的網(wǎng)絡負載達到瓶頸，I/O時間延長。

表1不同Benchmark下工作節(jié)點數(shù)與效率值

Benchmark不同節(jié)點數(shù)下的執(zhí)行效率/%直接運行123BenchmarkA10082.3878.5469.11BenchmarkB10087.5783.6576.96BenchmarkC10086.9682.9076.67

表2描述了研究分片時間與分布式系統(tǒng)工作效率的關(guān)系，從結(jié)果顯示，分片時間對于工作效率的影響是比較小的。分片時間決定了子任務的任務切分粒度，當分片時間比較短的時候，任務粒度較小，處理時間比較穩(wěn)定，但是分片過多引起I/O效率有下降的趨勢。當分片時間比較大的時候，任務粒度過大，有可能導致有些工作節(jié)點的閑置，導致計算框架整體效率下降。總體來說，分片時間與分布式計算框架的效率關(guān)系并不是太大，選擇2～20 s的分片大小均可以取得一個相對穩(wěn)定的工作效率。

表2分片時間改變時效率執(zhí)行時間的變化

節(jié)點不用分片時間下的執(zhí)行時間/s2s4s8s16s32s1節(jié)點121.26120.23120.49114.70119.392節(jié)點63.5962.8964.0660.6069.453節(jié)點48.1848.4147.5147.9760.134節(jié)點38.6938.2537.9038.8944.45

表3描述了TaskTracker使用本地存儲和使用遠端存儲對于單節(jié)點效率的影響。可以看到，無論使用本地節(jié)點還是遠端節(jié)點，都有著節(jié)點數(shù)增大，效率降低的問題，同時由于I/O時間的增加，使用遠程緩存的工作效率會下降7個百分點左右。所以使用迭代的策略是很有必要的。

表3使用公有存儲和私有存儲對效率值的影響

存儲方式不同節(jié)點數(shù)下的執(zhí)行效率/%1234私有存儲86.9682.9076.6774.98公有存儲80.9176.5569.5767.04

5小結(jié)

本文提出了一種基于Celery的分布式視頻處理框架，該框架借鑒了Hadoop的架構(gòu)設計，并對進行了優(yōu)化使之能夠兼容計算資源與儲存資源的橫向擴展和新的計算應用的縱向擴展。此外，本文還提出了對于計算框架的一些優(yōu)化機制，實驗數(shù)據(jù)證明分布式計算框架能夠高效地利用多節(jié)點實現(xiàn)視頻計算。

參考文獻：

[1]DIAZ-SANCHEZ D，MARIN-LOPEZ A， ALMENAREZ F， et al. A distributed transcoding system for mobile video delivery[C]// 2012 5th Joint IFIP Wireless and Mobile Networking Conference (WMNC).[S.l.]：IEEE， 2012： 10-16.

[2]YOO D， SIM K M. A comparative review of job scheduling for MapReduce[C]//2011 IEEE International Conference on Cloud Computing and Intelligence Systems (CCIS).[S.l.]：IEEE，2011： 353-358.

[3]YE X， HUANG M， ZHU D， et al. A novel blocks placement strategy for Hadoop[C]//2012 IEEE/ACIS 11th International Conference on Computer and Information Science. [S.l.]：IEEE， 2012： 3-7.

[4]陳珍. 基于 MapReduce 的海量視頻轉(zhuǎn)碼系統(tǒng)優(yōu)化機制[D]. 武漢：華中科技大學， 2013.

Distributed video processing system based on Celery

HUO Da，SONG Li

(ImageCommunicationandNetworkEngineeringInstitute，ElectronicEngineering，ShanghaiJiaoTongUniversity，Shanghai200240，China)

Abstract:In view of using computer cluster to solve video processing problem， a distributed video processing system based on celery is proposed in this paper. Using Hadoop’s structure as a reference， a three-tier structure is presented， including API server layer， JobTracker layer and TaskTracker layer. This structure enables the system to be compatible with both computing resource and functional extension. In this paper， structure of the system is presented in detail together with optimization. The experimental data indicate that the system is proved to be efficient with multiple calculating nodes.

Key words:distributed computing；cloud computing；transcoding

中圖分類號：TP338.8

文獻標志碼:A

DOI：10.16280/j.videoe.2016.04.003

基金項目：國家自然科學基金項目(61221001)

作者簡介：

霍達，碩士生，主要研究方向為視頻傳輸協(xié)議和分布式視頻處理框架；

宋利，副教授，研究方向為圖像處理、視頻編碼。

責任編輯：時雯

收稿日期：2015-12-15

文獻引用格式：霍達，宋利. 基于Celery的分布式視頻計算處理框架[J].電視技術(shù)，2016，40(4)：12-17.

HUO D，SONG L. Distributed video processing system based on Celery [J].Video engineering，2016，40(4)：12-17.