, , ,,

(1.上海民用飛機健康監(jiān)控工程技術(shù)研究中心，上海 200241； 2.北京航天測控技術(shù)有限公司，北京 100041； 3.北京市高速交通工具智能診斷與健康管理重點實驗室，北京 100041；4.軌道交通裝備全壽命周期狀態(tài)監(jiān)測與智能管理技術(shù)與應(yīng)用北京市工程實驗室，北京 100041)

0 引言

隨著計算機和測控技術(shù)的發(fā)展，試飛試驗數(shù)據(jù)呈現(xiàn)出參數(shù)多(上萬個)、數(shù)據(jù)量大(上百TB)、參數(shù)類型多樣化等特點[1]。飛行試驗數(shù)據(jù)是完成新機定型、鑒定的主要依據(jù)，同時也是支撐航空科技發(fā)展的寶貴資料。能否將大量飛行試驗數(shù)據(jù)有效管理并使用起來，這對我國航空事業(yè)的發(fā)展具有重要的現(xiàn)實意義。

飛行試驗是在真實飛行條件下進行的科學研究和產(chǎn)品試驗。它是航空科技發(fā)展的重要手段，是航空產(chǎn)品研制和鑒定的必須環(huán)節(jié)，是為用戶摸索和積累經(jīng)驗的有效途徑。試飛任務(wù)中的數(shù)據(jù)信息管理是一個龐大而復(fù)雜的系統(tǒng)工程，目前國內(nèi)傳統(tǒng)的試飛數(shù)據(jù)處理模式是：一型飛機、一個團隊、一套數(shù)據(jù)格式及處理程序，即由專業(yè)人員使用自行開發(fā)的軟件，對測試數(shù)據(jù)進行處理，中間結(jié)果及最終報告數(shù)據(jù)以操作系統(tǒng)文件的形式保存；有關(guān)數(shù)據(jù)處理的文檔人工處理。這種處理模式的缺點是：1)數(shù)據(jù)和程序的繼承性差，共享困難；2)低水平重復(fù)，不利于積累以往的經(jīng)驗；3)處理效率低，容易出錯；4)處理周期長，不利于快速做出下一步的試飛決策。

本文研究基于Spark的飛機試驗數(shù)據(jù)處理技術(shù)，將改變原有的“特定飛機，專用軟件”的研制模式，將復(fù)雜、繁瑣的數(shù)據(jù)管理工作抽象出來，使工程師能夠?qū)⒏嗟木械綌?shù)據(jù)的處理和分析工作中去。一方面，提煉共同要求，統(tǒng)一飛行試驗信息化標準，完成系統(tǒng)的統(tǒng)一規(guī)劃與部署，避免試飛信息化工作出現(xiàn)效率低下和資源浪費等問題。另一方面，通過建立健全和完善型號信息系統(tǒng)、數(shù)據(jù)與信息以及數(shù)據(jù)處理軟件的標準和規(guī)范，提高信息系統(tǒng)和軟件的可重復(fù)性，避免低水平的重復(fù)開發(fā)。

1 SPARK內(nèi)存計算框架

目前，應(yīng)用最為廣泛的大數(shù)據(jù)技術(shù)是hadoop以及其分布式架構(gòu)map-reduce。Hadoop MapReduce采用Master/slave 結(jié)構(gòu)。只要按照其編程規(guī)范，只需要編寫少量的業(yè)務(wù)邏輯代碼即可實現(xiàn)一個強大的海量數(shù)據(jù)并發(fā)處理程序。其核心思想是分而治之。Mapper負責把一個復(fù)雜的業(yè)務(wù)，任務(wù)分成若干個簡單的任務(wù)分發(fā)到網(wǎng)絡(luò)上的每個節(jié)點并行執(zhí)行，產(chǎn)生的結(jié)果由Reduce進行匯總，輸出到HDFS中，大大縮短了數(shù)據(jù)處理的時間開銷。MapReduce以一種可靠且容錯的方式對大規(guī)模集群海量數(shù)據(jù)進行數(shù)據(jù)處理，數(shù)據(jù)挖掘，機器學習等方面的操作[2-7]。

盡管該架構(gòu)具備開源分布式的特點且應(yīng)用范圍廣泛，但在大量數(shù)據(jù)離線計算過程中，map-reduce存在著大量的硬盤讀寫，這造成計算效率很低。Spark是用于大數(shù)據(jù)處理快速而通用的引擎，其采用分布式內(nèi)存計算方式，將過程數(shù)據(jù)保存在內(nèi)存中，減少了由于磁盤交互產(chǎn)生的I/O，從而提高數(shù)據(jù)計算效率[8-10]。

圖1 Hadoop和Spark的邏輯回歸

Spark擁有Hadoop MapReduce所具有的優(yōu)點，但和MapReduce的最大不同之處在于Spark是基于內(nèi)存的迭代式計算[11]。Spark的Job處理的中間輸出結(jié)果可以保存在內(nèi)存中，從而不再需要讀寫HDFS，除此之外，Map-Reduce在計算中只有兩個階段，即map和reduce。而在Spark的計算模型中，可以分為n階段，因為它內(nèi)存迭代式的，我們在處理完一個階段以后，可以繼續(xù)往下處理很多個階段，而不只是兩個階段。因此，Spark能更好地適用于數(shù)據(jù)挖掘與機器學習等需要迭代的MapReduce的算法。不僅實現(xiàn)了MapReduce的算子map函數(shù)和reduce函數(shù)及計算模型，還提供更為豐富的算子，如filter、join、groupByKey等，是一個用來實現(xiàn)快速而同用的集群計算的平臺。

此外，Spark應(yīng)用程序還離不開SparkContext和Executor。Executor負責執(zhí)行任務(wù)，運行Executor的機器稱為Worker節(jié)點，SparkContext由用戶程序啟動，通過資源調(diào)度模塊實現(xiàn)與Executor通信。圖2為Spark的操作機制。

圖2 Spark操作機制

集群管理器主要控制整個集群，監(jiān)控各工作節(jié)點。工作節(jié)點負責控制計算節(jié)點，啟動執(zhí)行器和驅(qū)動程序。執(zhí)行器是應(yīng)用程序運行在工作節(jié)點上的一個進程。

RDD(Resilient Distributed Datasets)[1],彈性分布式數(shù)據(jù)集，是分布式內(nèi)存的一個抽象概念，RDD提供了一種高度受限的共享內(nèi)存模型，即RDD是只讀的記錄分區(qū)的集合，只能通過在其他RDD執(zhí)行確定的轉(zhuǎn)換操作(如map、join和group by)而創(chuàng)建，然而這些限制使得實現(xiàn)容錯的開銷很低。對開發(fā)者而言，RDD可以看作是Spark的一個對象，它本身運行于內(nèi)存中，如讀文件是一個RDD，對文件計算是一個RDD，結(jié)果集也是一個RDD，不同的分片、數(shù)據(jù)之間的依賴、key-value類型的map數(shù)據(jù)都可以看做RDD。圖3顯示Spark計算的RDD模型。

圖3 Spark計算過程不同階段的RDD模型

2 海量飛行數(shù)據(jù)預(yù)處理方法

飛機試驗數(shù)據(jù)解析效率是影響大數(shù)據(jù)分析，挖掘額外信息資源的重要影響因素。隨著飛機集成度的提高，傳感器數(shù)據(jù)的增多，飛機試驗數(shù)據(jù)的參數(shù)種類與數(shù)據(jù)量劇增。傳統(tǒng)的飛機數(shù)據(jù)解析方法主要是針對源碼數(shù)據(jù)量小的情況，無法滿足當前大型飛機試驗任務(wù)的需求。而針對大規(guī)模源碼數(shù)據(jù)，目前一般都是工程師對任務(wù)進行分工，每個人負責任務(wù)的一部分，在各自的機器上運行，最后將結(jié)果進行匯總，這不僅會增加人力成本，也會增加解析的復(fù)雜度，降低解析效率，影響下一步試飛決策。因此，需要建立一種針對大規(guī)模飛行試驗數(shù)據(jù)的預(yù)處理方法。

本文一方面采用Spark處理框架，通過分布式計算方式，減少磁盤交互產(chǎn)生的I/O，提高計算效率；另一方面，利用飛機數(shù)據(jù)采集特點，減少不必要的解析，節(jié)省集群內(nèi)存計算空間，提高解析效率。從數(shù)據(jù)解析過程優(yōu)化與并行處理的角度，提出了基于Spark的海量飛行試驗數(shù)據(jù)的預(yù)處理方法。

2.1 飛行數(shù)據(jù)解析優(yōu)化

飛機運行過程中，機載采集器根據(jù)規(guī)定的采樣率對機載數(shù)據(jù)進行采集，當次試飛結(jié)束后下載原始數(shù)據(jù)。由于數(shù)據(jù)參數(shù)格式固定，在數(shù)據(jù)采集過程，根據(jù)實際情況解析前段數(shù)據(jù)，直到獲取所有需要的參數(shù)初始信息位置，包括參數(shù)的采樣初始時間，整個數(shù)據(jù)文件采樣的初始時間以及數(shù)據(jù)文件出現(xiàn)的參數(shù)名全集等。

確定數(shù)據(jù)文件中存在的參數(shù)集合，預(yù)先剔除參數(shù)組中不存在于當前數(shù)據(jù)文件中的參數(shù)群，只提取存在于數(shù)據(jù)文件中的參數(shù)屬性，有效地跳過不必要的解析過程。同時減少廣播變量(Broadcast)占用的內(nèi)存，為集群內(nèi)存計算節(jié)省空間，顯著提高解析的效率。

進而，切割并行數(shù)據(jù)，生成若干個原始數(shù)據(jù)包。切包的規(guī)則是只能在兩個單包之間切，而不能跨包，否則會丟失數(shù)據(jù)，通過不斷計算單包的長度來用指針偏移的方法來切包。

在此基礎(chǔ)上，對飛行測試數(shù)據(jù)進行預(yù)處理。預(yù)處理過程分為兩個階段，配置階段和應(yīng)用階段。圖4顯示了兩個階段的過程，包括控制流程和數(shù)據(jù)流程。

圖4 圖解碼階段的數(shù)據(jù)預(yù)處理步驟

在配置階段，主要是在試飛數(shù)據(jù)處理設(shè)備正式投入運行之前和運行過程中，進行系統(tǒng)基礎(chǔ)信息的配置與擴展開發(fā)，并且進行系統(tǒng)數(shù)據(jù)庫的管理，為系統(tǒng)提供原始數(shù)據(jù)的解碼、分析邏輯、算法功能等方面的配置和擴展開發(fā)能力，使系統(tǒng)具備持續(xù)的完善能力。

1)首先進行的是用戶配置、試驗信息配置與基礎(chǔ)信息配置的工作。

2)其次是完成對配置文件的解析和試驗數(shù)據(jù)的預(yù)處理。

3)再次是存儲原始二進制數(shù)據(jù)和解析后工程值數(shù)據(jù)。

4)最后針對解析后數(shù)據(jù)為用戶提供數(shù)據(jù)服務(wù)，包括數(shù)據(jù)的查詢、下載、導(dǎo)出、分析、二次計算等。

在準備階段，從配置文件中讀取幀格式的配置信息，包括幀參數(shù)的位置，采樣率，通道數(shù)等。然后讀取參數(shù)數(shù)據(jù)表中所有幀參數(shù)的數(shù)據(jù)類型、分辨率、數(shù)據(jù)長度等。最后，根據(jù)參數(shù)的位置信息設(shè)置參數(shù)索引表。

在解碼階段，當接收到源代碼時，根據(jù)429、232、664協(xié)議規(guī)定的幀格式對幀頭進行解析，獲取當前幀的索引表唯一標識，通過標識確定當前幀采集的參數(shù)索引表，進而根據(jù)參數(shù)信息將源碼數(shù)據(jù)解析為工程值數(shù)據(jù)。

圖5 解碼階段的數(shù)據(jù)預(yù)處理步驟

2.2 基于SPARK的數(shù)據(jù)處理技術(shù)

在使用SPARK技術(shù)來解決大規(guī)模數(shù)據(jù)預(yù)處理問題的過程中，如何在每個預(yù)處理步驟中構(gòu)建測試數(shù)據(jù)RDD模型以及如何使用SPARK算子至關(guān)重要，直接影響數(shù)據(jù)預(yù)處理的效率。圖6為基于SPARK的數(shù)據(jù)處理流程圖。

圖6 基于SPARK的數(shù)據(jù)處理流程圖

基于SPARK的數(shù)據(jù)處理具體步驟如下：

步驟1：初始化。包括共享變量初始化，通過解析配置文件，獲取參數(shù)信息和協(xié)議解析規(guī)則，并打包成廣播變量作為集群共享信息，同時初始化累加器，作為分布式運算過程中不同線程之間數(shù)據(jù)通信的媒介；以及待解析數(shù)據(jù)初始化，將源碼數(shù)據(jù)文件的上傳到HDFS，并按照1 GB一個單元進行預(yù)先切包。最后還對spark環(huán)境進行初始化，指定基本的參數(shù)配置，保證spark的高效運行。

步驟2：分布式切包。將上傳到HDFS上的數(shù)據(jù)加載到分布式內(nèi)存中，在加載的過程中，同時將數(shù)據(jù)每隔N(一般可以設(shè)定為1000)個幀切一次包，并將所有切后的包重組為K(一般可以設(shè)定為2000)份，設(shè)定為每份啟動一個線程，進行分布式處理。將加載到分布式算子中。一部分數(shù)據(jù)將打開一個線程。創(chuàng)建數(shù)據(jù)結(jié)構(gòu)為JavaRDD>的RDD數(shù)據(jù)集，該數(shù)據(jù)集作為分布式處理的數(shù)據(jù)源，其中每個byte數(shù)組代表的是幀二進制數(shù)據(jù)。

步驟3：解析單包。根據(jù)步驟一中初始化的廣播變量信息，解析幀頭，獲取幀頭的標識信息，根據(jù)標識以及解析的配置文件獲取當前幀采集的參數(shù)以及參數(shù)屬性，進而將源碼幀數(shù)據(jù)解析成工程值數(shù)據(jù)。期間，待解析的參數(shù)組范圍之外的參數(shù)直接跳過，保證解析速度的最大化；此外，整個解析過程通過mapPartitionsToPair算子執(zhí)行并行處理，各個線程共享廣播變量，盡可能的避免不同線程在運算過程中的shuffle操作，減少線程之間由于通信導(dǎo)致的延遲。該步驟獲取的數(shù)據(jù)格式為JavaPairRDD，存儲的內(nèi)容為<參數(shù)組+時間，參數(shù)位置+參數(shù)值>信息，保證每個基本參數(shù)都從二進制解析成工程值數(shù)據(jù)。

步驟4： 參數(shù)值聚合。對步驟三得到的結(jié)果進行聚合操作，執(zhí)行g(shù)roupBykey算子，將key(參數(shù)組+時間)相同的value(參數(shù)位置+參數(shù)值)聚合在一起，得到當前參數(shù)組當前幀時間的所有參數(shù)數(shù)據(jù)集，并將數(shù)據(jù)集根據(jù)參數(shù)組規(guī)定的順序組成長字符串，以便獲取同一采樣時間點的參數(shù)值，便于進行事后試驗數(shù)據(jù)的關(guān)聯(lián)分析。最終獲取到的數(shù)據(jù)結(jié)構(gòu)為JavaPairRDD>，存儲的結(jié)果為<參數(shù)組+時間，參數(shù)值字符串集合>。

圖7 主機配置

步驟5：求解合成參數(shù)。有一些參數(shù)需要一些基本參數(shù)經(jīng)過一定的運算法則轉(zhuǎn)換才能獲取，如兩個參數(shù)的加減乘除運算、位權(quán)重運算、EU轉(zhuǎn)換、多項式計算等。所以在將基礎(chǔ)參數(shù)工程值解析出來后，需要找到合成參數(shù)需要的基本參數(shù)相同時間點的數(shù)據(jù)集，然后經(jīng)過算法運行，得到合成參數(shù)的值。其結(jié)果形式與步驟5產(chǎn)生的結(jié)果一致。

步驟6：過濾數(shù)據(jù)。針對不同的參數(shù)組，需要分別導(dǎo)出到不同的文件中，以保證不同試驗科目的獨立性，便于不同專業(yè)的專家對數(shù)據(jù)進行分析。然后以參數(shù)組+時間字符串中的參數(shù)組為過濾條件，使用filter算子，篩選出所需的參數(shù)組數(shù)據(jù)集，由于執(zhí)行的是filter算子，得到的RDD數(shù)據(jù)集數(shù)據(jù)結(jié)構(gòu)沒有變化，結(jié)果形式與步驟5一致。

步驟7：首先需要對整體數(shù)據(jù)進行全局排序，以保證所有數(shù)據(jù)能夠以時間順序進行排列。此時數(shù)據(jù)是分布式的散播在內(nèi)存中，為了將參數(shù)組數(shù)據(jù)以規(guī)定的數(shù)據(jù)格式導(dǎo)出，需要將內(nèi)存中所有的相關(guān)數(shù)據(jù)合并在一起，使用repartitionAndSortWithinPartitions算子，將partitioner參數(shù)設(shè)置為1，即得到合并后的數(shù)據(jù)。由于只是數(shù)據(jù)的合并，所以RDD結(jié)構(gòu)仍然不變，為JavaPairRDD。

步驟8：保存數(shù)據(jù)文件。 使用saveAsTextFile算子，將步驟7得到的合并結(jié)果保存到HDFS的文件系統(tǒng)中，每個參數(shù)組都分別形成獨立的文件系統(tǒng)，通過服務(wù)的形式分發(fā)給不同試驗科目的專業(yè)人員，進行進一步的數(shù)據(jù)分析。

步驟9：轉(zhuǎn)化為HBase字段。預(yù)處理的數(shù)據(jù)需要進行卸載，一方面以系統(tǒng)文件的形式分發(fā)給專業(yè)人員，另一方面需要持久化存儲到分布式數(shù)據(jù)庫HBase中，以便時候的隨時查詢和進一步的分析，通過flatMapToPair算子將步驟6的RDD數(shù)據(jù)集轉(zhuǎn)化為滿足HBase特定形式的JavaPairRDD 。

步驟10：保存HFile文件。由于實際產(chǎn)生的數(shù)據(jù)量過大，直接存儲到HBase容易出現(xiàn)內(nèi)存溢出問題，并且消耗時間較長。故而采取先轉(zhuǎn)存為HFile文件的形式進行HBase存儲，使用saveAsNewAPIHadoopFile算子將數(shù)據(jù)集轉(zhuǎn)化為HFile文件。

步驟11：存儲到HBase。通過doBulkLoad方法將HFile文件轉(zhuǎn)化為HBase數(shù)據(jù)，進而完成送源碼數(shù)據(jù)到工程值數(shù)據(jù)的處理工作。

3 實驗驗證

3.1 實驗環(huán)境

實驗環(huán)境部署在有3個物理服務(wù)器的集群中，其中一個服務(wù)器既作為主管理節(jié)點，又作為計算節(jié)點；另一個服務(wù)器作為備管理節(jié)點和計算節(jié)點；最后一個服務(wù)器作為計算節(jié)點，具體環(huán)境配置如表1所示。

表1 實驗環(huán)境配置

試驗數(shù)據(jù)來源于某型號飛機試飛過程產(chǎn)生的數(shù)據(jù)，數(shù)據(jù)量為30 GB。將試驗數(shù)據(jù)分成三份，分別在三臺主機上進行預(yù)處理。

3.2 實驗結(jié)果分析

基于Spark的飛機試驗數(shù)據(jù)預(yù)處理結(jié)果，如圖8所示。數(shù)據(jù)切包、單包解析、參數(shù)聚合、參數(shù)合成、數(shù)據(jù)存儲所需時間約為30分鐘。傳統(tǒng)數(shù)據(jù)處理平臺數(shù)據(jù)預(yù)處理時間則需要9小時。相較于傳統(tǒng)數(shù)據(jù)處理方法，本文提出的數(shù)據(jù)預(yù)處理方法大大提高了解析效率，克服了傳統(tǒng)數(shù)據(jù)處理方式效率低下，單個機器內(nèi)存和CPU等硬件條件不足的問題。

圖8 各階段處理時間

主機一運行線程333個，各線程累積執(zhí)行時間2.3 h，解析工程數(shù)據(jù)文件9.5 GB；主機二運行線程331個，各線程累積執(zhí)行時間2.3 h，解析工程數(shù)據(jù)文件9.5 GB；主機三運行線程336個，各線程累積執(zhí)行時間2.3 h，解析工程數(shù)據(jù)文件9.7 GB。各線程執(zhí)行時間最小為5 s，最大為55 s。數(shù)據(jù)垃圾回收時間最小為0 ms，最大為32 s。各線程處理原始數(shù)據(jù)最小為3.6 MB，最大為11.3 MB，解析數(shù)據(jù)輸出大小最小為17.2 MB，最大為42.2 MB，如圖9所示。盡管本文提出的方法相較于傳統(tǒng)方法，大大提高了數(shù)據(jù)解析效率。但各線程負載不均衡，未充分利用系統(tǒng)資源，影響系統(tǒng)運行性能，對數(shù)據(jù)解析時間產(chǎn)生一定的負面影響，這也是后續(xù)需要進一步改進的地方。

圖9 飛行測試數(shù)據(jù)預(yù)處理記錄

4 結(jié)論

本文針對海量飛行數(shù)據(jù)預(yù)處理方法處理效率低、處理周期長等問題，提出了基于Spark的飛機試驗數(shù)據(jù)預(yù)處理方法，并以某型飛機30 GB的試飛數(shù)據(jù)為例進行實驗驗證，結(jié)果顯示本文提出的數(shù)據(jù)預(yù)處理方法相較于傳統(tǒng)的大規(guī)模數(shù)據(jù)預(yù)處理方法可有效地縮短數(shù)據(jù)預(yù)處理時間，提高處理效率，幫助用戶快速做出下一步試飛決策。未來將重點研究數(shù)據(jù)傾斜調(diào)優(yōu)與Shuffle調(diào)優(yōu)的問題，優(yōu)化Spark的性能，進一步提高數(shù)據(jù)解析效率。