基于Flink的空間科學(xué)衛(wèi)星數(shù)據(jù)實(shí)時(shí)處理方法

高立京,陳志敏,王春梅,王 靜

(1. 中國(guó)科學(xué)院國(guó)家空間科學(xué)中心,北京 100190;2. 中國(guó)科學(xué)院大學(xué)計(jì)算機(jī)科學(xué)與技術(shù)學(xué)院,北京 100049)

1 引言

空間科學(xué)是一門研究發(fā)生在日地空間、行星際空間乃至整個(gè)宇宙空間的物理、天文、化學(xué)及生命等自然現(xiàn)象及其規(guī)律的學(xué)科[1],航天器是研究其具體內(nèi)容的主要平臺(tái)。在各種航天器的種類中,空間科學(xué)衛(wèi)星是一類以執(zhí)行特定科學(xué)探測(cè)任務(wù)為總體目標(biāo)、以獲取科學(xué)探測(cè)數(shù)據(jù)為主要需求的航天器。

空間科學(xué)衛(wèi)星下行數(shù)據(jù)主要分為工程參數(shù)數(shù)據(jù)、衛(wèi)星平臺(tái)數(shù)據(jù)和科學(xué)數(shù)據(jù)。其中,工程參數(shù)數(shù)據(jù)和衛(wèi)星平臺(tái)數(shù)據(jù)用于地面科研人員判斷有效載荷和衛(wèi)星的運(yùn)行狀態(tài)正常與否,地面數(shù)據(jù)處理系統(tǒng)應(yīng)能夠盡可能低延遲地處理工程參數(shù)數(shù)據(jù)和衛(wèi)星平臺(tái)數(shù)據(jù),從而幫助地面科研人員第一時(shí)間發(fā)現(xiàn)有效載荷或衛(wèi)星的異常并給出相應(yīng)的解決方案,減少甚至避免因載荷或衛(wèi)星的異常而帶來(lái)的損失;空間科學(xué)衛(wèi)星下行數(shù)據(jù)中的科學(xué)數(shù)據(jù)是星上有效載荷探測(cè)到的數(shù)據(jù),這些科學(xué)數(shù)據(jù)對(duì)于地面科研人員開展相應(yīng)的科學(xué)研究工作有著至關(guān)重要的作用,由于空間科學(xué)衛(wèi)星上攜帶的有效載荷數(shù)量較多,并且有效載荷的科學(xué)數(shù)據(jù)采樣頻率極高,導(dǎo)致產(chǎn)生的科學(xué)數(shù)據(jù)數(shù)據(jù)量巨大。因此研究如何高吞吐率、低延遲地實(shí)時(shí)處理空間科學(xué)衛(wèi)星下行數(shù)據(jù)是很有必要的。

目前實(shí)時(shí)處理空間科學(xué)衛(wèi)星下行數(shù)據(jù)的方法主要分為兩種:流式計(jì)算處理方法和非流式計(jì)算處理方法。在非流式計(jì)算方法中,文獻(xiàn)[1]設(shè)計(jì)兩層聯(lián)合索引,將數(shù)據(jù)處理問題轉(zhuǎn)化為處理索引表和源包數(shù)據(jù)單元的問題,提高了空間科學(xué)衛(wèi)星下行數(shù)據(jù)處理的效率;文獻(xiàn)[2]借鑒MapReduce的思想采用單機(jī)多線程的方式處理空間科學(xué)衛(wèi)星下行數(shù)據(jù),同時(shí)將哈希算法與歸并算法結(jié)合在其中,實(shí)現(xiàn)了較高的數(shù)據(jù)處理吞吐率。在流式計(jì)算方法中,文獻(xiàn)[3]采用Spark Streaming流式計(jì)算框架的方式處理空間科學(xué)衛(wèi)星下行數(shù)據(jù),實(shí)現(xiàn)了較高的數(shù)據(jù)吞吐率;文獻(xiàn)[4]重新定義數(shù)據(jù)處理程序的數(shù)據(jù)結(jié)構(gòu),將其與Storm流式計(jì)算框架結(jié)合實(shí)現(xiàn)并行處理,相較于原有系統(tǒng)極大地提高了數(shù)據(jù)處理的吞吐率。

主流的流式計(jì)算框架有Strom、Spark Streaming和Flink,在實(shí)時(shí)處理空間科學(xué)衛(wèi)星下行數(shù)據(jù)的領(lǐng)域中,基于Storm和Spark Streaming的數(shù)據(jù)處理方式均取得了不錯(cuò)的性能表現(xiàn)[3,4],但基于Flink流式計(jì)算框架的空間科學(xué)衛(wèi)星下行數(shù)據(jù)的處理方式尚未有人涉足。鑒于以上調(diào)研情況,本文提出一種基于Flink流式計(jì)算框架的空間科學(xué)衛(wèi)星下行數(shù)據(jù)的實(shí)時(shí)處理方法,并討論了在不同并行度時(shí)該方法在吞吐率和延遲方面的性能表現(xiàn)。

2 空間科學(xué)衛(wèi)星下行數(shù)據(jù)

引力波暴高能電磁對(duì)應(yīng)體全天監(jiān)測(cè)器衛(wèi)星(Gravitational wave high-energy Electromagnetic Counterpart All-sky Monitor,GECAM)是用來(lái)探測(cè)引力波事件所發(fā)出的高能光的空間科學(xué)衛(wèi)星[5],本文以GECAM為例說(shuō)明本文中處理數(shù)據(jù)的數(shù)據(jù)結(jié)構(gòu)。GECAM的下行數(shù)據(jù)遵循空間數(shù)據(jù)系統(tǒng)咨詢委員會(huì)(Consultative Committee for Space Data Systems, CCSDS)提出的高級(jí)在軌系統(tǒng)(Advanced Orbit System, AOS)標(biāo)準(zhǔn)[6],并且按照“AOS傳輸幀”和“源包”兩層的結(jié)構(gòu)分層組織,本文中處理的數(shù)據(jù)為“內(nèi)部數(shù)據(jù)幀”,是由“源包”處理得到的。從“AOS傳輸幀”到“源包”再到“內(nèi)部數(shù)據(jù)幀”的具體處理流程如圖1所示。

圖1 AOS傳輸幀的處理流程

“AOS傳輸幀”的格式如表1所示,其經(jīng)過以下處理步驟后可得到“源包”:

表1 AOS傳輸幀格式

1)幀同步:從衛(wèi)星下行的原始碼流中逐位查找同步碼,從而確定每個(gè)AOS傳輸幀的起始位置和結(jié)束位置。

2)解擾:加擾的數(shù)據(jù)是除同步碼以外的數(shù)據(jù),以AOS傳輸幀為處理單位,對(duì)這部分?jǐn)?shù)據(jù)進(jìn)行位運(yùn)算得到加擾前的數(shù)據(jù)。

3)RS譯碼:以“AOS傳輸幀”為處理單位,使用RS譯碼對(duì)衛(wèi)星下行數(shù)據(jù)在傳輸過程中產(chǎn)生的誤碼進(jìn)行糾正。

4)虛擬信道分離:以“AOS傳輸幀”為處理單位,解析幀頭的“虛擬信道標(biāo)識(shí)符”字段信息,按照虛擬信道標(biāo)識(shí)符(Virtual Channel Identifier, VCID)提取不同信道的源包,比如實(shí)時(shí)工程參數(shù)源包、實(shí)時(shí)科學(xué)數(shù)據(jù)源包等。

“源包”是一種面向應(yīng)用且長(zhǎng)度可變的數(shù)據(jù)格式,它使用應(yīng)用過程標(biāo)識(shí)符(Application Identifier, APID)對(duì)衛(wèi)星在采集和傳輸過程中的不同的數(shù)據(jù)源設(shè)備進(jìn)行區(qū)分[4]?！霸窗钡母袷饺绫?所示,其經(jīng)過以下處理步驟后可得到“內(nèi)部數(shù)據(jù)幀”:

表2 源包格式

1)源包同步:對(duì)于從“AOS傳輸幀”解析而來(lái)的碼流逐位查找同步碼,從而確定每個(gè)“源包”的起始位置和結(jié)束位置。

2)源包分包:以“源包”為處理單位,解析“源包”頭部的“應(yīng)用過程標(biāo)識(shí)符”字段信息,按照應(yīng)用過程標(biāo)識(shí)符(Application Identifier, APID)提取不同有效載荷數(shù)據(jù)的源包。

3)物理量反演:以“源包”為處理單位,根據(jù)源包的“應(yīng)用過程標(biāo)識(shí)符”字段所對(duì)應(yīng)的特定公式來(lái)計(jì)算編碼所對(duì)應(yīng)的物理量數(shù)據(jù)。

4)添加幀頭:以源包為處理單位,添加相關(guān)幀頭信息。

“內(nèi)部數(shù)據(jù)幀”的格式如表3所示,本文中涉及到的數(shù)據(jù)處理就是要解析“內(nèi)部數(shù)據(jù)幀”中的各個(gè)字段的十六進(jìn)制編碼,將該編碼與配置文件中的相關(guān)配置信息做映射,將其解析為具體的物理量信息。

表3 內(nèi)部數(shù)據(jù)幀格式

3 基于Flink的數(shù)據(jù)處理方法

為了提高數(shù)據(jù)處理的并行度,借助消息中間件Kafka在數(shù)據(jù)進(jìn)入Flink集群前將數(shù)據(jù)進(jìn)行分流。Kafka是Apache軟件基金會(huì)開源的一個(gè)多分區(qū)、多副本、基于發(fā)布訂閱模式的分布式消息中間件,與傳統(tǒng)的消息系統(tǒng)相比,Kafka有著系統(tǒng)解耦、緩沖、異步通信、擴(kuò)展性強(qiáng)、可恢復(fù)性等特點(diǎn)。Kafka作為大數(shù)據(jù)的組件,能夠與HBase、Flink、Spark等開源大數(shù)據(jù)框架無(wú)縫集成,在大數(shù)據(jù)的應(yīng)用領(lǐng)域,Kafka多被用作數(shù)據(jù)流實(shí)時(shí)傳輸?shù)墓艿繹7]。Kafka的體系結(jié)構(gòu)如圖2所示,Kafka主要由以下幾個(gè)部分組成[7,9]:

圖2 Kafka體系結(jié)構(gòu)圖[8]

1)Topic(主題):Topic在邏輯上劃分消息的種類,同一個(gè)Topic中的消息可以看作是同一種業(yè)務(wù)類型的消息集合。本文中設(shè)置Topic A用于存放解析前的“內(nèi)部數(shù)據(jù)幀”,設(shè)置Topic B用于存放“內(nèi)部數(shù)據(jù)幀”解析后的“結(jié)果幀”。

2)Partition(分區(qū)):在同一個(gè)Topic下可以劃分多個(gè)Partition,并且這多個(gè)Partition中的消息是不重復(fù)的,這樣的設(shè)計(jì)可以使得多個(gè)Partition中的消息并行地被消費(fèi),從而可以提高Kafka的吞吐率。同一個(gè)Topic下的Partition會(huì)分配到不同的Broker中,所以可以通過增加Partition的數(shù)量實(shí)現(xiàn)水平擴(kuò)展。本文中保證Topic A和Topic B中Partition的數(shù)量一直相等。

3)Broker(服務(wù)代理節(jié)點(diǎn)):Broker是Kafka的一個(gè)獨(dú)立服務(wù)節(jié)點(diǎn),一個(gè)或多個(gè)Broker組成Kafka集群。

4)Producer(生產(chǎn)者):Producer負(fù)責(zé)創(chuàng)建消息并將消息投遞到Broker中,Broker便會(huì)根據(jù)預(yù)設(shè)規(guī)則將該消息存入對(duì)應(yīng)的Topic下的某個(gè)Partition中。在本文中,針對(duì)于Topic A方面,Producer為UDP服務(wù)端,投遞的消息為解析前的“內(nèi)部數(shù)據(jù)幀”;針對(duì)于Topic B方面,Producer為Flink集群的各個(gè)節(jié)點(diǎn),投遞的消息為處理后的“結(jié)果幀”。

5)Consumer (消費(fèi)者):Consumer訂閱指定的Topic,根據(jù)預(yù)設(shè)規(guī)則主動(dòng)從Broker中拉取到相關(guān)的消息。在本文中,針對(duì)于Topic A方面,Consumer為Flink集群的各個(gè)節(jié)點(diǎn),消費(fèi)的消息為處理前的“內(nèi)部數(shù)據(jù)幀”;針對(duì)于Topic B方面,Consumer為后處理模塊,消費(fèi)的消息為處理后的“結(jié)果幀”。

Flink是Apache軟件基金會(huì)開源的分布式實(shí)時(shí)流式計(jì)算框架,其核心是一個(gè)流數(shù)據(jù)的處理引擎,Flink將所有任務(wù)當(dāng)做流來(lái)處理[10]。Flink的體系結(jié)構(gòu)如圖3所示,Flink集群中主要包含以下幾種組件[10,11]:

圖3 Flink體系結(jié)構(gòu)圖[11]

1)Client(客戶端):Client負(fù)責(zé)向JobManager提交作業(yè)任務(wù),當(dāng)用戶提交了一個(gè)Flink應(yīng)用的時(shí)候,Client就會(huì)啟動(dòng)并將作業(yè)任務(wù)移交給JobManager。

2)JobManager(作業(yè)管理器):JobManager作為Flink集群中的主節(jié)點(diǎn),負(fù)責(zé)接收Client提交的作業(yè)任務(wù),并根據(jù)一定規(guī)則將任務(wù)分配給TaskManager執(zhí)行,同時(shí)接收來(lái)自于TaskManager反饋的任務(wù)狀態(tài)和統(tǒng)計(jì)信息,然后將這些信息反饋給Client。

3)TaskManager(任務(wù)管理器):TaskManager作為Flink集群中的從節(jié)點(diǎn),負(fù)責(zé)接收J(rèn)obManager所分配的作業(yè)任務(wù),對(duì)作業(yè)任務(wù)中的數(shù)據(jù)流執(zhí)行具體的業(yè)務(wù)邏輯處理,同時(shí)TaskManager會(huì)將任務(wù)的狀態(tài)和統(tǒng)計(jì)信息反饋給JobManager。在本文中,TaskManager負(fù)責(zé)的具體業(yè)務(wù)邏輯為根據(jù)配置文件的相關(guān)配置信息解析“內(nèi)部數(shù)據(jù)幀”中各個(gè)字段的十六進(jìn)制編碼。

4)Slot(插槽):TaskManager中可以根據(jù)硬件配置指定Slot的數(shù)量,具體地,Slot的最大設(shè)定數(shù)量等于Flink集群中所有CPU的物理核心總數(shù)量,Slot的數(shù)量決定了TaskManager中運(yùn)行的數(shù)據(jù)處理程序的最大并行度。在TaskManager中運(yùn)行的每個(gè)并行的數(shù)據(jù)處理程序都分別占用一個(gè)Slot資源。

系統(tǒng)整體的架構(gòu)如圖4所示,整體的數(shù)據(jù)處理流程描述如下。

1)系統(tǒng)整體的數(shù)據(jù)來(lái)源方為上游的參數(shù)處理與綜合判讀系統(tǒng),上游系統(tǒng)通過UDP組播的形式將滿足表3格式的“內(nèi)部數(shù)據(jù)幀”實(shí)時(shí)數(shù)據(jù)流發(fā)送給UDP服務(wù)器端,UDP服務(wù)器端在接收到該實(shí)時(shí)數(shù)據(jù)流后進(jìn)行合法性校驗(yàn),將合法的“內(nèi)部數(shù)據(jù)幀”存入Kafka中。Kafka同一個(gè)Topic下的多個(gè)Partition中存放的數(shù)據(jù)是不重復(fù)的[9],借助這個(gè)機(jī)制,在“內(nèi)部數(shù)據(jù)幀”進(jìn)入Flink集群之前,可以使用Kafka進(jìn)行數(shù)據(jù)的分流,從而提升數(shù)據(jù)處理的并行度。對(duì)數(shù)據(jù)幀進(jìn)行分流時(shí),將數(shù)據(jù)幀按照到來(lái)的順序依次存入Partition-1, Partition-2…Partition-(4n-1), Partition-4n中,并重復(fù)這個(gè)存入過程。

2)為了保證能夠做到高吞吐率、低延遲地處理數(shù)據(jù),本文采用Flink集群的方式來(lái)解析“內(nèi)部數(shù)據(jù)幀”,設(shè)置Flink集群中數(shù)據(jù)處理程序的并行度等于Kafka的Topic A和Topic B中Partition的數(shù)量,這樣可以保證每一個(gè)Slot中的數(shù)據(jù)處理程序并行地分別去Topic A中各自序號(hào)對(duì)應(yīng)的Partition中拉取“內(nèi)部數(shù)據(jù)幀”,然后并行地解析“內(nèi)部數(shù)據(jù)幀”,在解析完畢后并行地將解析得到的“結(jié)果幀”存入到Topic B中各自序號(hào)對(duì)應(yīng)的Partition中。

3)后處理模塊從Kafka的Topic B中,依次獲取Partition-1, Partition-2…Partition-(4n-1), Partiton-4n中解析好的“結(jié)果幀”,然后將“結(jié)果幀”進(jìn)行合并,恢復(fù)其原有的順序,最后將“結(jié)果幀”投遞到下游系統(tǒng)即數(shù)據(jù)處理與任務(wù)監(jiān)視系統(tǒng)中。

4 仿真分析驗(yàn)證

本文采用仿真驗(yàn)證基于Flink流式計(jì)算框架的空間科學(xué)衛(wèi)星下行數(shù)據(jù)實(shí)時(shí)處理方法。實(shí)驗(yàn)數(shù)據(jù)來(lái)源于GECAM空間科學(xué)衛(wèi)星在執(zhí)行科學(xué)任務(wù)過程中下行的真實(shí)數(shù)據(jù),該數(shù)據(jù)遵循AOS格式標(biāo)準(zhǔn),且按照“AOS傳輸幀”和“源包”兩層的結(jié)構(gòu)分層組織,“AOS傳輸幀”的格式如表1所示,“源包”的格式如表2所示,該數(shù)據(jù)經(jīng)過圖1中的處理步驟后,得到525170條滿足表3格式的“內(nèi)部數(shù)據(jù)幀”,幀的長(zhǎng)度為不等長(zhǎng)。編程語(yǔ)言采用Java,使用的編輯器為IntelliJ IDEA Community 2021。使用3臺(tái)服務(wù)器節(jié)點(diǎn)搭建集群實(shí)驗(yàn)環(huán)境,3臺(tái)服務(wù)器的配置信息如表4所示。其中主節(jié)點(diǎn)運(yùn)行的為Flink集群中的JobManager進(jìn)程,負(fù)責(zé)協(xié)調(diào)作業(yè)任務(wù)在各個(gè)節(jié)點(diǎn)上的分布式執(zhí)行;2個(gè)從節(jié)點(diǎn)分別運(yùn)行的為Flink集群中的TaskManager進(jìn)程,負(fù)責(zé)執(zhí)行具體的作業(yè)任務(wù)即解析“內(nèi)部數(shù)據(jù)幀”。3臺(tái)服務(wù)器節(jié)點(diǎn)通過交換機(jī)建立局域網(wǎng)連接,3臺(tái)服務(wù)器節(jié)點(diǎn)的硬件架構(gòu)如圖5所示。

表4 3臺(tái)服務(wù)器節(jié)點(diǎn)配置信息

圖5 3臺(tái)服務(wù)器節(jié)點(diǎn)硬件架構(gòu)圖

在硬件設(shè)備和數(shù)據(jù)處理程序均一致時(shí),影響Flink集群處理數(shù)據(jù)的性能的關(guān)鍵點(diǎn)在于其處理數(shù)據(jù)的并行度。本文中使用的兩個(gè)服務(wù)器從節(jié)點(diǎn)均搭載4核心的CPU,故在每一個(gè)服務(wù)器節(jié)點(diǎn)中可以設(shè)置Slot的值為4,這說(shuō)明該Flink集群可以為在其中運(yùn)行的數(shù)據(jù)處理程序提供到的最大并行度為8。在數(shù)據(jù)流進(jìn)入到Flink集群之前,本文使用到了消息中間件Kafka進(jìn)行數(shù)據(jù)分流,考慮到發(fā)揮并行處理的最大作用,這里設(shè)置Kafka中Topic A的Partition數(shù)量和Topic B的Partition數(shù)量均等于Flink集群中數(shù)據(jù)處理程序的并行度的數(shù)量。舉個(gè)例子,在Topic A中,Partition的數(shù)量設(shè)置為3,則在Flink集群中,數(shù)據(jù)處理程序的并行度設(shè)置為3,并且在Topic B中,Partition的數(shù)量也設(shè)置為3。

在本文的仿真中,將Kafka中Partition的數(shù)量與Flink集群中數(shù)據(jù)處理程序的并行度兩者的組合稱為配置,因本文中的Flink集群可以為在其中運(yùn)行的數(shù)據(jù)處理程序提供從1到8的并行度,所以本文討論了從“1-1”到“8-8”多種配置下的空間科學(xué)衛(wèi)星下行數(shù)據(jù)處理的延遲和吞吐率,其中“1-1”表示Kafka的Topic A和Topic B的Partition數(shù)量為1,Flink中數(shù)據(jù)處理程序的并行度為1。在不同配置下的延遲統(tǒng)計(jì)信息如圖6所示,圖中的橫坐標(biāo)表示配置,縱坐標(biāo)表示數(shù)據(jù)處理的延遲,單位是毫秒(ms),并且對(duì)于每一種配置均做了5次實(shí)驗(yàn)取平均值,從而得到了在每一種配置下數(shù)據(jù)處理的最大延遲、最小延遲和平均延遲。從圖6中可以看出,隨著數(shù)據(jù)處理的并行度的提高,數(shù)據(jù)處理的最大延遲、最小延遲和平均延遲總體上均呈現(xiàn)下降的趨勢(shì),在配置為“8-8”時(shí),平均延遲達(dá)到了最低值112ms,相較于配置為“1-1”時(shí),平均延遲降低了78.75%。在不同配置下的吞吐率統(tǒng)計(jì)信息如圖7所示,圖中的橫坐標(biāo)表示配置,縱坐標(biāo)表示數(shù)據(jù)處理的吞吐率,單位是兆比特每秒(Mbps),同樣的對(duì)于每一種配置均做了5次實(shí)驗(yàn)取平均值,從而得到了在每一種配置下數(shù)據(jù)處理的吞吐率。從圖7中可以看出,隨著數(shù)據(jù)處理并行度的提高,數(shù)據(jù)處理的吞吐率逐漸增大,在配置為“8-8”時(shí),吞吐率達(dá)到了峰值417.49Mbps,相較于配置為“1-1”時(shí),吞吐率提高了86.60%。同時(shí),結(jié)合圖6和圖7可以看出,在總體趨勢(shì)上增加并行度可以提高吞吐率降低延遲,由此可以推斷,通過增加集群中服務(wù)器的數(shù)量即增加集群中CPU的總核數(shù)來(lái)提高數(shù)據(jù)處理程序的并行度,可以實(shí)現(xiàn)在合理的范圍內(nèi)進(jìn)一步提高數(shù)據(jù)處理的吞吐率、降低數(shù)據(jù)處理的延遲。

圖6 不同配置下的延遲統(tǒng)計(jì)

圖7 不同配置下的吞吐率統(tǒng)計(jì)

5 結(jié)束語(yǔ)

本文設(shè)計(jì)并實(shí)現(xiàn)了一種基于Flink流式計(jì)算框架的空間科學(xué)衛(wèi)星下行數(shù)據(jù)實(shí)時(shí)處理方法,該方法借助消息中間件Kafka各分區(qū)消息不重復(fù)的機(jī)制,將空間科學(xué)衛(wèi)星下行數(shù)據(jù)依次存入不同分區(qū)中實(shí)現(xiàn)數(shù)據(jù)分流,搭建Flink集群用于高速處理數(shù)據(jù),使分流后的數(shù)據(jù)依次流入到Flink集群的各個(gè)并行處理單元中實(shí)現(xiàn)并行處理。仿真結(jié)果表明,本文提出的方法具有較高的吞吐率和較低的延遲,在并行度為8時(shí),吞吐率可以達(dá)到417.49Mbps,平均延遲可以達(dá)到112ms,并且可以通過水平擴(kuò)展服務(wù)器的數(shù)量從而實(shí)現(xiàn)更高的吞吐率和更低的延遲,可以滿足地面數(shù)據(jù)處理系統(tǒng)高速實(shí)時(shí)處理空間科學(xué)衛(wèi)星下行數(shù)據(jù)的需求。