張文彬，王春梅，王 靜，陳 托，智 佳

(1.中國科學院 國家空間科學中心，北京 100190；2.中國科學院大學 計算機科學與技術(shù)學院，北京 100049)

0 引 言

科學衛(wèi)星有效載荷產(chǎn)生的科學探測數(shù)據(jù)具有數(shù)據(jù)量大、參數(shù)多、處理實時性要求高的特點，其中參數(shù)解析是有效載荷數(shù)據(jù)實時處理的關(guān)鍵環(huán)節(jié)，其參數(shù)越多數(shù)據(jù)量越大，解析過程就越復雜越耗時。目前有效載荷數(shù)據(jù)的實時參數(shù)解析主要采用單機多線程處理方法[1]，其存在吞吐率低、擴展能力弱的不足，因此，提高有效載荷參數(shù)解析的速率具有必要性。

當前主流的大數(shù)據(jù)流式計算框架Storm、Spark Streaming等具有低延遲、高吞吐、可擴展等優(yōu)勢[2,3]，本文結(jié)合衛(wèi)星有效載荷數(shù)據(jù)流的特點[4]，利用大數(shù)據(jù)計算框架良好的實時處理性能和易擴展的能力，以提高有效載荷參數(shù)解析的吞吐率[5]。其中Spark[6,7]提供的生態(tài)系統(tǒng)具備同時支持批處理、交互式查詢和流數(shù)據(jù)處理的優(yōu)勢，可實現(xiàn)數(shù)據(jù)的無縫共享。Spark Streaming是Spark計算引擎內(nèi)的流式計算框架，因此，本文結(jié)合Spark Streaming和Kafka[8,9]，設(shè)計并實現(xiàn)了一種有效載荷實時參數(shù)解析的處理方法，以提高有效載荷數(shù)據(jù)參數(shù)解析處理的實時性。

1 有效載荷參數(shù)解析的數(shù)據(jù)源

有效載荷數(shù)傳數(shù)據(jù)，其格式遵循國際空間數(shù)據(jù)系統(tǒng)咨詢委員會(consultative committee for space data systems，CCSDS)的高級在軌系統(tǒng)(advanced orbit system，AOS)標準[10]，有效載荷數(shù)傳數(shù)據(jù)結(jié)構(gòu)見表1。

有效載荷數(shù)傳數(shù)據(jù)的處理[11,12]步驟如圖1所示，在對CCSDS格式的數(shù)傳/遙測數(shù)據(jù)進行AOS幀同步、解密、解擾、分包、拼接等預(yù)處理之后，形成中間數(shù)據(jù)格式，稱為數(shù)據(jù)幀，其主要結(jié)構(gòu)見表2，其中，數(shù)據(jù)段部分存放各類參數(shù)的二進制編碼，參數(shù)解析過程即是針對數(shù)據(jù)段中的各個參數(shù)編碼進行處理，將其按要求解析為電流、溫度等物理量。

表1 有效載荷數(shù)傳數(shù)據(jù)結(jié)構(gòu)

圖1 數(shù)據(jù)處理流程

傳輸標簽衛(wèi)星標識消息長度幀計數(shù)衛(wèi)星時間碼狀態(tài)量計數(shù)2B2B4B6B1B2B數(shù)據(jù)段應(yīng)用數(shù)據(jù)變長

2 參數(shù)解析處理方法

為提高有效載荷數(shù)據(jù)的實時參數(shù)解析吞吐率，采用基于Spark Streaming與Kafka相結(jié)合的方法，處理流程如圖2所示，利用Kafka集群作為消息中間件實現(xiàn)數(shù)據(jù)分流，為數(shù)據(jù)接入提供保障，流式計算部分采用Spark Streaming集群作為計算平臺，通過Spark Streaming獲取Kafka消息隊列的數(shù)據(jù)[13]，并對參數(shù)進行解析，然后將解析結(jié)果發(fā)送給Kafka作為數(shù)據(jù)緩沖區(qū)進行合并，最終將計算結(jié)果發(fā)送給實現(xiàn)參數(shù)錄入的軟件。

圖2 系統(tǒng)處理流程

為避免數(shù)據(jù)源產(chǎn)生堆積，提高數(shù)據(jù)處理的速率，采用Kafka集群實現(xiàn)數(shù)據(jù)分流。Kafka是一種基于發(fā)布/訂閱的分布式消息系統(tǒng)，可以在多個分布式生產(chǎn)者、消費者并發(fā)的情況下，保證消息的有序性和負載均衡，可同時支持離線數(shù)據(jù)和實時數(shù)據(jù)的處理，其吞吐量可隨集群的擴展而線性增加，且消息持久化的時間復雜度為O(1)[14]，具有高吞吐率、高可靠性和易擴展的優(yōu)點。

Kafka集群部署模式如圖3所示，在Kafka集群部署架構(gòu)中，可以存在多個Producer(生產(chǎn)者)，生產(chǎn)者負責收集消息并將消息發(fā)布到Broker(代理)相應(yīng)的Topic(主題)中，Broker接收消息，并將消息在本地持久化，數(shù)據(jù)按照Topic名存儲在不同分類中，一個Topic可以分成多個Partition(分區(qū))，每個Partition內(nèi)部消息強有序，將數(shù)據(jù)處理為多個分區(qū)的消息隊列流，用以作為中間數(shù)據(jù)源，在隊列底端存在多個Consumer(消費者)[15]。消費者是消息的真正使用者，從Topic中讀取隊列消息進行處理[16]。其中Broker1、Broker2分別部署在不同服務(wù)器上，Spark Streaming的實時計算程序充當消費者訂閱Topic1，當Topic1中有數(shù)據(jù)，會將數(shù)據(jù)不停的從集群的指定消息隊列中發(fā)送給消費者做參數(shù)解析處理。

圖3 Kafka集群部署模式

Spark Streaming是在Spark架構(gòu)上基于離散化數(shù)據(jù)流(discretized stream，DStream)模型擴展的分布式流式計算框架，其中，DStream表示持續(xù)不斷的數(shù)據(jù)流，其可以是不同類型數(shù)據(jù)源的數(shù)據(jù)，包括文件流、套接字流、基于Kafka的輸入數(shù)據(jù)流等[17]。Spark Streaming可以在多達100個節(jié)點上運行，實時處理吞吐率能達到秒級的延遲需求[18]，可以有效實現(xiàn)高吞吐的參數(shù)解析處理，并且Spark Streaming支持節(jié)點的錯誤恢復，是具備容錯機制的實時流數(shù)據(jù)的處理框架[19]。

因此，采用Spark Streaming實現(xiàn)有效載荷數(shù)據(jù)幀的實時參數(shù)解析，搭建Spark Streaming集群，設(shè)置集群中的主從節(jié)點，其中主節(jié)點負責資源分配，從節(jié)點負責監(jiān)控本節(jié)點的CPU及內(nèi)存情況，接收主節(jié)點命令。將Spark Strea-ming 作為消費者訂閱Kafka集群中的Topic1，當Topic1中有數(shù)據(jù)時，消費者從消息隊列中獲取數(shù)據(jù)。Spark Strea-ming 處理進程在獲取數(shù)據(jù)之后按照參數(shù)名、起止位置、轉(zhuǎn)換公式等結(jié)構(gòu)信息，對參數(shù)進行解析，其處理架構(gòu)如圖4所示，主要包括如下步驟。

(1)數(shù)據(jù)分流

利用Kafka將數(shù)據(jù)源轉(zhuǎn)換為消息隊列流，按FIFO(first input first output，先進先出)方式有序緩存于 Topic1 中的各個Partition中，等待Spark處理進程作為消費者消費數(shù)據(jù)。

(2)Spark處理進程

主節(jié)點為從節(jié)點的Executor(執(zhí)行進程)分配內(nèi)存、CPU內(nèi)核等資源，并啟動Executor進程，每個從節(jié)點運行若干Executor，每個Executor獨立運行參數(shù)解析處理程序，即將數(shù)據(jù)幀按照數(shù)據(jù)格式中的參數(shù)名、起止位置、轉(zhuǎn)換公式等信息進行解析，解析結(jié)果作為Producer發(fā)送給Kafka的Topic2。

(3)歸并

將Topic2中各個Partition的參數(shù)結(jié)果按時間先后順序進行歸并。Kafka的Partition內(nèi)部消息強有序，從各Partition中獲取結(jié)果數(shù)據(jù)1～n，按時間順序排序?qū)?shù)解析結(jié)果合并，最后將參數(shù)存入數(shù)據(jù)庫。

3 方法實現(xiàn)與仿真驗證

搭建3臺虛擬機組成Spark Streaming集群，集群配置見表3。

測試采用表2格式的仿真數(shù)據(jù)進行實驗，對單機多線程方法與基于Spark Streaming的集群方法進行了仿真測試，令單機多線程方法運行在表3中的任一臺從節(jié)點(Worker1或Worker2)虛擬機上，基于Spark Streaming的集群方法運行在表3中的3臺虛擬機搭建的集群上，測試結(jié)果見表4，結(jié)果表明在相同的處理單元數(shù)量下，單機多線程處理方法的吞吐率為10.24 Mbps，基于Spark Streaming的集群方法為25.56 Mbps，相比單機多線程方法的數(shù)據(jù)處理吞吐率提高了150%，并且基于Spark Streaming的集群方法可以通過增加從節(jié)點的方式進一步提升處理速率，具有很強的擴展能力，在實時參數(shù)解析處理中更具優(yōu)勢。

4 結(jié)束語

本文提出并實現(xiàn)了一種基于Spark的有效載荷實時參數(shù)解析處理方法，采用了Apache Kafka和Spark Streaming相結(jié)合的處理方法，利用Kafka對有效載荷實時數(shù)據(jù)分流，Spark Streaming獲取數(shù)據(jù)并進行實時參數(shù)解析，解決了單機多線程方法在吞吐率和擴展能力上的局限性，提高了有效載荷參數(shù)解析處理的實時吞吐率，仿真結(jié)果表明，所提方法相比目前的單機多線程方法在相同處理單元配置下的數(shù)據(jù)吞吐率提高了150%，具有更優(yōu)的實時參數(shù)解析能力。

圖4 參數(shù)解析處理架構(gòu)

序號名稱角色配置1Master主節(jié)點CPU:2.83 GHz,2核心;內(nèi)存:6 G;硬盤:20 GB;操作系統(tǒng):CentOS72Worker1從節(jié)點CPU:2.83 GHz,4核心;內(nèi)存:4 G;硬盤:20 GB;操作系統(tǒng):CentOS73Worker2從節(jié)點CPU:2.83 GHz,4核心;內(nèi)存:4 G;硬盤:20 GB;操作系統(tǒng):CentOS7

表4 兩種方法測試結(jié)果比較