基于Spark Streaming的氣象自動站實時流處理與存儲系統

馬 彬，李玉濤，許 琪

(1.江蘇省氣象信息中心，江蘇 南京 210005；2.江蘇省氣侯中心，江蘇 南京 210005)

0 引 言

隨著氣象觀測設備信息化程度的大幅提高，地面自動氣象站的時空密度不斷增加，氣象自動站已實現了分鐘加密觀測，產生的觀測數據量也呈指數級增長[1-3]，具有數據種類多、數據規模大、數據實時性強以及價值密度低等特點[4]。同時，自動站也已成為監視天氣變化、決策服務輔助支持的重要手段，為氣象預報、氣象防災減災、氣候預測與生態環境評估等提供十分重要的基礎數據支撐[5-6]。為進一步推動氣象事業高質量發展，中國氣象局提出以氣象信息化推動氣象現代化的發展戰略，而作為氣象服務最為核心業務應用之一的自動站數據，也將面臨重大的挑戰，對數據的實時采集處理、數據質量、數據存儲及大規模查詢等要求也越來越高，需要在秒級甚至更短時間內完成數據的全流程處理，從而提高響應效率，發揮更重要的應用價值。

Spark作為主流的開源分布式計算框架，具有可擴展、高吞吐量和可容錯等特點[7]。Spark Streaming則是Spark框架的實時流處理組件[8]，采用了一種新的離散流處理模型，進行計算處理時，將數據流以時間片為單位進行切割形成彈性分布式數據集RDD(Resilient Distributed Dataset)，而RDD提供了共享內存式的并行運算，因此Spark在批處理、迭代計算、交互式查詢和流處理等多種計算模式方面具備高時效的處理能力[9]。HBase是Hadoop Database的簡稱，屬于NoSQL[10-11]，是一個KeyValue類型的分布式存儲數據庫，具有海量存儲、高并發、高可靠以及可伸縮等特點[12]，適應于氣象數據的存儲管理應用，能夠有效地解決氣象大數據的存儲和檢索響應不足等問題[13]。

該文重點研究了基于Spark Streaming的氣象自動站數據實時流處理與分布式存儲應用技術，通過對實時氣壓、溫度、降水、風速、濕度等氣象要素的基本質控算法設計，實現了自動站數據流式采集、解碼、基本質控和入庫的分布式實時數據全流程處理功能。

1 現狀與需求分析

目前，現有的氣象自動站數據經基層臺站設備采集后，通過寬帶網統一傳輸至省級落地入庫，通過氣象資料業務系統對數據進行質控后再共享或分發給各類業務應用，其傳輸流程環節多，易導致數據處理不及時、交互響應慢、統計時效差等問題。此外，由于省級部署的一些決策、服務等業務平臺開發早，大都依托于傳統的關系型數據庫進行海量數據的存儲，在大批量多并發查詢情況下，通常存在著數據檢索能力不足、檢索性能下降等問題。并且省級開發的業務系統大都應用比較深入，覆蓋業務范圍廣，升級改造存在一定困難，極大地影響了氣象數據的應用服務體驗。因此，為滿足大規模自動站數據能夠在秒級完成與用戶的交互響應，就對數據處理流程節點的集約化設計和高質量、高可靠的數據存儲及檢索功能提出了更高的要求。

Spark Streaming技術的應用研究，為實現高效實時數據處理提供了技術支撐[14]。開展基于數據流式采集、傳輸、質控、存儲為一體的氣象自動站數據全流程設計及應用研究，能夠從根本上解決氣象自動站數據落地環節多、任務處理耦合緊、處理系統部署分散等問題；進一步提升氣象大數據的傳輸效率與數據質量；改善業務平臺在多并發，長序列數據檢索使用時的問題。在當前氣象信息化任務高效推動的背景下，對氣象數據以流式實現全流程的處理具有迫切的應用研究需求。

2 系統總體設計

2.1 系統總體架構設計

系統總體架構主要分為4層：數據層、處理層、邏輯層和應用層，如圖1所示。

圖1 系統總體架構

數據層主要分為源數據和處理后的分布式數據存儲，源數據通過處理層相應的Spark Streaming組件實現數據的實時流處理功能，Flume為源數據文件的解碼采集模塊，獲取源數據流后將數據暫存到Kafka消息中間件，由Spark Streaming調用Kafka中的數據做實時處理，再將所需處理的數據根據邏輯層相應的算法及表結構設計通過接口存儲到分布式數據庫中。邏輯層設計的功能主要包括溫度、壓強、降水等氣象自動站數據基本要素的查詢以及檢索。應用層分為用戶應用和業務應用兩類，用戶應用指為科研人員提供數據服務，主要為客戶端及科研用的虛擬化服務器等，業務應用則指根據各業務場景的應用需求開發的平臺和系統等。

2.2 系統技術分析

2.2.1 數據實時采集技術分析

Flume[15]是一個分布式、高可靠、具備可定制化能力的日志采集傳輸系統[16]，其數據流由事件(Event)貫穿始終，Event代表一個完整數據的最小單元，是事務的基本單位。這些Event攜帶日志數據并且帶有頭信息，由Agent外部的Source通過特定的格式化后生成，然后再發送到指定的目的地(Sink)進行下一步操作。為確保數據能夠成功傳輸，通常Source會把Event推送至一個緩沖區(Channel)中，待確保前一個Event已由Sink處理完后，Channel再清空自己的緩存數據。Sink則負責持久化日志或者把事件推送到外部其他的Source。

Flume以Agent為最小的獨立運行單位，每個Agent由Source、Channel和Sink組件構成，Flume結構如圖2所示。

圖2 Flume Agent結構

Source：負責接收數據或通過特定機制生成數據，然后以Flume的Event格式傳遞給一個或者多個Channel，Flume提供多種數據接收的方式，但Source必須至少和一個Channel相關聯。

Channel：位于Source和Sink間的一種存儲容器，用于緩存Source推送進來的數據，起著鏈接橋梁的作用。Channel將從Source接收到的Event緩存起來，直到它們被Sink消費完成。同時，它支持一個完整的事務，可提供順序保證，這樣就確保了數據在收發時的一致性，并且可以和任意數量的Source和Sink工作。

Sink：負責將數據傳輸到下一跳或最終目的地，任務成功結束后將數據從Channel移除。典型的Sink類型為：存儲數據到目的的終端Sink，如HDFS、HBase；自動消耗的Sinks，如Null Sink；以及用于Agent間通信的IPC Sink，如Avro。

Event：Flume數據傳輸的基本單元。一行文本內容會被反序列化成一個Event。

2.2.2 數據實時處理技術分析

數據的實時處理主要是對采集的氣象自動站臟數據進行清洗。數據格式錯誤、數值錯誤等多種原因導致的臟數據，若不經過清洗就直接解析傳入到架構組件中，會產生極大的成本和時間代價[17]。數據實時處理依據數據實時傳輸運行框架環境，采用Kafka和Spark Streaming實現。Kafka是一個分布式、高吞吐、基于發布訂閱的消息系統[18]，具有持久化、高吞吐、分布式、多客戶端支持以及實時等特點，適用于離線和在線的消息消費。利用Kafka技術可在廉價的PC Server上搭建起大規模消息系統，從而大幅提升數據實時處理能力，其結構如圖3所示。

圖3 Kafka結構

Broker：在Kafka集群上一個服務器稱為一個Broker。

Topic：每條發布到Kafka集群的消息都有一個類別，稱為Topic。

Consumer：向Topic訂閱，并且接受發布到這些Topic的消息。

Producer：負責發布消息到Kafka Broker。

數據與處理：主要對數據的合法性進行檢測，包括界限值、奇異值，以及數據內部一致性等，該文主要檢測各字段是否與氣象行業數據質控體系相符，如要素數值、日期等格式。

2.2.3 數據分布式存儲技術分析

經過Spark Streaming 流式處理后的數據將存儲到HBase[19]。HBase的實現包括三個主要功能組件，即庫函數，一個Master主服務器和許多個RegionServer。主服務器Master負責管理和維護HBase表的分區信息，維護RegionServer列表、分配Region和負載均衡。RegionServer存儲和維護分配給自己的Region，用來處理來自客戶端的讀寫請求。客戶端則不依賴于Master，它是通過請求ZooKeeper獲取存儲了Region和RegionServer映射關系的元數據表信息，然后直接從RegionServer獲取數據。ZooKeeper是一個分布式應用程序協調服務[20-21]，提供統一命名服務、配置管理和分布式鎖等基礎服務[22]，具有集群管理、Master選舉、分布式協調通知和分布式隊列等功能[23]。基于以上所述的工作模式，實現了HBase快速響應的特點，其結構如圖4所示。

圖4 HBase結構

在HBase的使用中，RowKey的設計極為重要。HBase按照RowKey的順序去遍歷所有可能的數據，然后再依次匹配相應列的值，直到獲取所需的數據。若RowKey設計不合理，會造成單個Region訪問壓力過大，難以有效發揮其處理性能。因此，設計的RowKey要確保其具有唯一性，然后充分利用其有序性，均勻地分布在各個HBase節點上。

2.3 系統關鍵技術設計

2.3.1 數據實時采集設計

將自動站文件目錄設置為Flume監控目錄，Flume將收到的源文件實時解析為數據流發送到Spark Streaming的監控輸入目錄。Flume是分布式的，可以同步處理到達的多個文件，同時它也提供了許多可調的故障恢復和容錯機制，當某個節點出現故障時，數據能夠被傳送到其他節點上而不會丟失，從而保證數據的完整性。以下是源文件以氣壓要素進行解析數據流的代碼樣例。

static List p_list = new ArrayList();//創建氣壓數據緩存list

String[] tmp = null;

FileReader fr = new FileReader(FileName);

BufferedReader br = new BufferedReader(fr);

while ((line = br.readLine()) != null) {

tmp = line.split(" ");

for (int i = 0; i < tmp.length; i++) {

tmp_list.add(tmp[i]);

}

}

p_list.add(tmp_list.get(1));//獲取氣壓標識段所有元素數據

String[] p Tmp = null;

p Tmp = p_list.get(0).split(" ");

String station_P = p Tmp[1];//臺站氣壓數據

String sea_P = p Tmp[2];//海平面氣壓數據

PressureClass2 pc2 = new PressureClass2();

pc2.setStation_P(station_P);//封裝臺站氣壓數據

pc2.setSea_P(sea_P); //封裝海平面氣壓數據

2.3.2 數據實時處理設計

數據在前端由Flume收集起來，通過Kafka來做緩存和容災，最后由Spark Streaming來做實時處理。為減小代碼間的耦合性，設計中將自動站各要素的質控算法代碼單獨放入一個特定的Streaming來對數據流進行實時清洗，再將Kafka的Topic中清洗完成后的數據放入另一個Topic中供后續的業務來處理。相關代碼示意如下：

String topics = "weatherTopic";

SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("name");

JavaSparkContext sparkContext = new JavaSparkContext(conf);

sparkContext.setLogLevel("WARN");

JavaStreamingContext ssc = new JavaStreamingContext(sparkContext, Durations.seconds(1));//初始化StringContext

Collection topicsSet = new HashSet<>(Arrays.asList(topics.split(",")));

Map kafkaParams = new HashMap<>();

kafkaParams.put("metadata.broker.list", "ip:port");//消費端設置

kafkaParams.put("bootstrap.servers", " ip:port ");

2.3.3 數據質控算法設計

氣象自動站源文件數據要素多達100余種，設計中針對常用的基本氣象數據要素，如氣壓、溫度、降水、風速、濕度等，依據氣象行業標準質控技術規程中對質控對象的分類，進行流數據的基本質控算法設計。本質控算法設計分為分鐘數據質控和小時數據質控兩類，質控方法則選用基本的格式檢查法、界限值檢查法和內部一致性檢查法，相關規則如下所述。

規則1 在QC方法質控碼(QCcode)中，為每個QC方法設置7級質控碼，取值范圍為-3～3，其中0表示正確，±1表示可疑，±2表示警告，±3表示數據錯誤。QC碼的符號表示疑誤數據偏離真值的方向，負號表示疑誤數據偏小，正號表示疑誤數據偏大，即數據質量隨著控制碼數值絕對值的增加而降低。為方便使用，提出各質控碼的符號表示形式，用f(e)表示數據格式檢查法的QC碼。f1(e)表示界限值檢查法的QC碼，f2(e)表示范圍值檢查法的QC碼，f3(e)表示內部一致性檢查法的QC碼，其中e表示質控的要素，如P表示氣壓，T表示溫度，U表示相對濕度，F表示風速，R表示降水。

規則2 在數據格式檢查法中，按照地面自動站氣象要素資料(國家站)格式說明。本站氣壓、海平面氣壓、最高本站氣壓、最低本站氣壓字段長度均為5 Byte；臺站氣溫、最高氣溫、最低氣溫、分鐘相對濕度、最小相對濕度、2分鐘平均風速、10分鐘平均風速、最大風速、小時降水量、每1小時極大風速、過去6小時極大風速、過去12小時極大風速均為4 Byte。

規則3 在界限值檢查法中，各氣象要素的界限值閾值范圍參考氣象觀測規范進行設定，如 氣壓數據定義范圍為[500,1 200]，溫度為[-55,55]，濕度為[0,100]，小時降水為[0,600]，風速為[0,150]。

規則4 在內部一致性檢查法中，定義氣壓、溫度和相對濕度的第60分鐘的數據同小時正點數據不一致即為警告數據，即E59≠En。此外，對于同一文件內的數據，當前時刻的氣壓、溫度要素值應介于最小值與最大值之間，即Emin≤En≤Emax，但需提出的是，當前的溫度值(用Tn表示)應不小于當前的露點溫度值(用Td表示)，即Td≤Tn。對于特殊的要素，如相對濕度，其當前時刻的值不應小于最小相對濕度值，即Emin≤En，本定義中的E均表示質控的要素。具體算法說明如表1和表2所示。

表1 分鐘數據質控算法設計說明

表2 小時數據質控算法設計說明

2.3.4 數據分布式存儲表結構設計

在該系統數據庫的表結構設計中，需充分考慮源文件數據中所含基本信息要素的唯一性，利用MD5方法將行主鍵站號散列化，方便將所有數據散列到不同的Region上，從而有助于提高數據的查詢響應效率。因此將行主鍵設計采用MD5(站號)+源文件內觀測時間組合的方法存儲Spark Streaming實時質控后的要素數據，將質控后的數據存儲在一個定義為aws_qcalldata的列族中，具體設計如表3所示。

表3 自動站表結構

MD5算法是在MD4算法基礎上由美國密碼學家羅納德·李維斯特(Ronald Linn Rivest)設計，通過該算法能夠將任意長度的文本轉換為一個固定長度(128位)的散列值，成為一個不可逆的字符串，從而有效保證了數據的安全性和信息傳輸的完整性[24-26]。功能實現代碼樣例如下：

public static String getSaltMD5(String password) {

Random random = new Random();

StringBuilder sBuilder = new StringBuilder(16);

sBuilder.append(random.nextInt(99999999)).append(random.nextInt(99999999));

int len = sBuilder.length();

if (len < 16) {

for (int i = 0; i < 16 - len; i++) {

sBuilder.append("0");

}

}

String salt = sBuilder.toString();

password = md5Hex(password + salt);

char[] cs = new char[48];

for (int i = 0; i < 48; i += 3) {

cs[i] = password.charAt(i / 3 * 2);

char c = salt.charAt(i / 3);

cs[i + 1] = c;

cs[i + 2] = password.charAt(i / 3 * 2 + 1);

}

return String.valueOf(cs);

}

bc.setId(MD5Utils.getSaltMD5(stationNum)+ObservTime);//basic info 字段引入MD5算法

3 測試結果分析

3.1 系統測試運行環境

系統測試運行環境采用4臺虛擬化服務器做集群，其具體部署情況如表4所示。

表4 測試環境部署架構

系統運行環境基于Spark 2.1.3版本，在程序中引入，數據庫集群由MPPDB 6.5.1.5構建，Flume版本基于1.7.0，kafka為1.0.0，以及jdk 1.8。

3.2 數據流處理性能測試

江蘇共有國家基本氣象觀測自動站70余個，每個文件表示該臺站在某時刻所采集的氣象數據信息，文件內容包含了百余類氣象要素數據。測試中則采用實時業務應用的國家基本氣象觀測站共享目錄中文本文件作為源數據，將數據滾動復制到應用服務器Flume實時監聽目錄data/listen_data中，使用Spark Streaming將Kakfa中的流數據寫入MPPDB，針對單個源文件及多個源文件從目錄監聽、解析、實時質控至數據入庫的全流程耗時分別進行5次統計，取平均耗時作為測試結果，具體數據如表5和表6所示。

表5 單文件數據流處理性能測試結果 ms

分析表5和表6可知，單站文件從源文件解析至入庫全流程平均耗時在4 s以內；多個文件同時傳輸時，每個文件入庫的全流程平均耗時約1.46 s，能夠達到秒級數據處理能力，與現有的氣象自動站數據在1 min內到達預報員桌面的要求相比，完全滿足實時業務的應用需求。

表6 多文件數據流處理性能測試結果

3.3 多場景數據查詢性能測試

在多場景數據查詢性能測試中，根據常用的業務查詢需求，將多個場景的SQL寫入至一個查詢文件，同時對MPPDB數據庫的數據翻倍到TB量級后進行批量查詢，查三次取平均值。

測試結果如表7所示。

表7 多場景數據查詢性能測試結果

測試結果表明，在不同場景的查詢條件下，該系統的點查詢響應為毫秒級，加權查詢為秒級，能夠有效地支撐實時業務中對氣象自動站數據的查詢應用。

4 結束語

從業務實際應用需求出發，基于Spark Streaming的流式計算框架，開展以數據流式采集、傳輸、質控、存儲為一體的氣象自動站數據全流程設計及應用研究，通過模擬業務中使用的情景進行性能測試，驗證該系統的可行性和適用性。通過對測試結果的分析，表明該系統能夠有效地提升氣象自動站數據的實時處理和查詢能力，與現有的數據處理系統相比具有以下優點：

(1) 將自動站數據常用的氣象要素質控算法設計并融入流處理組件中，實現基于數據流式采集、傳輸、質控、存儲為一體的氣象自動站數據全流程處理功能，減少了數據落地處理節點，進一步提高了數據處理的時效性和可用性。

(2)實現數據加密算法設計并應用在分布式數據庫中，在提高數據存儲和檢索效率的同時，極大地提升了數據的安全性。

(3)系統的設計開發從實際業務應用需求出發，且部署靈活，可作為省級自動站質控數據存儲的實時備份，為氣象業務提供更加可靠的數據保障。

但此系統在應用研究中也存在一些不足，如搭建系統環境的虛擬計算資源比較缺乏，數據庫和應用服務器配置低于部署要求的最低配置標準。同時，在集群應用的規劃上，為節約資源，將主應用和中間件部署在同一臺虛擬服務器上，一定程度上都影響了數據實時處理的性能。在后續的應用研究中，將把文件所有氣象要素數據的實時質控處理納入設計工作，進一步完善數據質控算法，搭建資源充足的系統環境，優化Spark Streaming的作業動態調度配置以及調優系統集群各組件的一致性、容錯性和高可用性等，開展更為深入的業務應用研究。