基于Spark技術(shù)的氣象數(shù)據(jù)分析

劉麗景,劉力維,王 林,郝婉婷

(1.西安培華學(xué)院,西安 710125; 2.西安交通大學(xué),西安 710049; 3.西安市臨潼區(qū)氣象局,西安 710600)

0 引言

隨著天氣數(shù)據(jù)類型的不斷增多、信息量的增加,社會(huì)對(duì)氣象數(shù)據(jù)實(shí)時(shí)獲取的要求越來越高,跟蹤、統(tǒng)計(jì)及收集實(shí)時(shí)數(shù)據(jù)對(duì)于諸多行業(yè)、企業(yè)及個(gè)人來說都具有較高的實(shí)用價(jià)值[1]。對(duì)于大多數(shù)氣象數(shù)據(jù)應(yīng)用場(chǎng)景來說,建立高效的數(shù)據(jù)收集、傳輸、存儲(chǔ)及分析系統(tǒng)至關(guān)重要。使用現(xiàn)代化的傳感器、衛(wèi)星遙感技術(shù)及物聯(lián)網(wǎng)技術(shù),可實(shí)時(shí)地收集、傳輸及發(fā)布?xì)庀髷?shù)據(jù)。利用流處理技術(shù)(Apache Kafka、Spark Streaming和Apache Flink)可實(shí)現(xiàn)實(shí)時(shí)數(shù)據(jù)的處理及分析[2],基于這些技術(shù)可更好地應(yīng)對(duì)氣象數(shù)據(jù)的增多及信息量的增加,為各行各業(yè)提供更精準(zhǔn)的氣象信息,支持決策及應(yīng)對(duì)天氣變化。本研究主要基于Kafka和Spark技術(shù),實(shí)現(xiàn)對(duì)陜西省各城市的氣象數(shù)據(jù)分析與可視化[3],對(duì)相對(duì)濕度、溫度、空氣質(zhì)量、風(fēng)級(jí)等進(jìn)行多維度可視化,能夠支持分析預(yù)測(cè)未來14 d高低溫度變化、未來7 d氣候分布及風(fēng)級(jí)等。

系統(tǒng)主要流程包括實(shí)時(shí)氣象數(shù)據(jù)采集、數(shù)據(jù)清洗與預(yù)處理、數(shù)據(jù)處理與分析、結(jié)構(gòu)化數(shù)據(jù)存儲(chǔ)、數(shù)據(jù)可視化等部分,利用網(wǎng)絡(luò)爬蟲技術(shù)采集中央氣象臺(tái)的氣象數(shù)據(jù),將數(shù)據(jù)發(fā)送到Kafka消息隊(duì)列[4]中。使用Spark技術(shù)快速高效分析數(shù)據(jù),將分析得到的結(jié)果存儲(chǔ)在MySQL中,基于Spring Boot框架搭建web系統(tǒng)管理數(shù)據(jù)并運(yùn)用Echarts進(jìn)行數(shù)據(jù)分析及可視化展示,具體過程如圖1所示。

圖1 數(shù)據(jù)分析處理流程Fig.1 Data analysis process

圖2 氣象基本信息表E_R圖Fig.2 Table E_R chart of basic meteorological information

圖3 天氣狀況表E_R圖Fig.3 Weather table E_R chart

圖4 一天內(nèi)相對(duì)濕度曲線值Fig.4 Relative humidity curve on one day

圖5 一天內(nèi)溫度曲線值Fig.5 Temperature curve on one day

圖6 一天空氣質(zhì)量變化Fig.6 Air quality changes on one day

采集陜西省各城市氣象數(shù)據(jù),數(shù)據(jù)主要來源于中央氣象臺(tái)官方網(wǎng)站。數(shù)據(jù)采集流程具體為:獲取陜西省及省內(nèi)各城市的編碼,進(jìn)行記錄,分析各城市24 h整點(diǎn)天氣頁(yè)面,分析頁(yè)面結(jié)構(gòu)、頁(yè)面源代碼等,基于爬蟲技術(shù)爬取各城市24 h的整點(diǎn)天氣數(shù)據(jù),將數(shù)據(jù)打包成JSON格式的消息發(fā)送到Kafka隊(duì)列中?；谂老x技術(shù)快速獲取實(shí)時(shí)數(shù)據(jù),設(shè)置定時(shí)器功能,設(shè)定每60 min執(zhí)行一次爬蟲程序,以獲取準(zhǔn)確的實(shí)時(shí)數(shù)據(jù)。爬蟲程序?qū)⑴廊〉降臄?shù)據(jù)通過BeautifulSoup庫(kù)來解析數(shù)據(jù),轉(zhuǎn)換為python對(duì)象[5]。提取的數(shù)據(jù)主要包括氣溫、城市、日期、風(fēng)速、風(fēng)向、濕度、降水量、氣壓等,提取到的數(shù)據(jù)最終被轉(zhuǎn)換為JSON格式的字符串,以便在Kafka隊(duì)列中進(jìn)行傳輸。

1 Kafka數(shù)據(jù)傳輸

經(jīng)爬蟲采集到的氣象數(shù)據(jù)被封裝成JSON字符串后發(fā)送到Kafka中名為“weather_data”的Topic[6]。Kafka以分布式方式存儲(chǔ)數(shù)據(jù),數(shù)據(jù)被復(fù)制到多個(gè)Broker上,Broker接收并持久化發(fā)布到Topic的數(shù)據(jù),確保數(shù)據(jù)的高可用性及容錯(cuò)性?？蛻舳嗽讷@取氣象數(shù)據(jù)時(shí),使用Kafka的Consumer客戶端訂閱“weather_data”這個(gè)Topic,Kafka將數(shù)據(jù)傳輸給訂閱該Topic的客戶端,客戶端可以按照自己的速率處理數(shù)據(jù)。整個(gè)過程中,Kafka的分布式架構(gòu)和高吞吐量特性確保了可靠、高效的氣象數(shù)據(jù)傳輸,保證數(shù)據(jù)能夠準(zhǔn)時(shí)送達(dá)消費(fèi)者,在數(shù)據(jù)量增大時(shí)也能夠處理并存儲(chǔ)大規(guī)模的氣象數(shù)據(jù)[7]。

2 數(shù)據(jù)清洗與預(yù)處理

數(shù)據(jù)清洗和預(yù)處理是在對(duì)數(shù)據(jù)進(jìn)行審查及校驗(yàn)過程中發(fā)現(xiàn)并糾正數(shù)據(jù)文件中可識(shí)別的錯(cuò)誤,按照一定的規(guī)則糾正錯(cuò)誤或沖突數(shù)據(jù)。通用的數(shù)據(jù)清洗與預(yù)處理主要包括處理異常數(shù)據(jù)、重復(fù)數(shù)值及缺失值等。

1)異常數(shù)值。采集的數(shù)據(jù)往往存在一些異常值,這些異常值可能由于數(shù)據(jù)采集錯(cuò)誤、傳輸錯(cuò)誤等原因?qū)е?會(huì)嚴(yán)重影響數(shù)據(jù)分析結(jié)果,因此需要進(jìn)行異常值處理。常用的異常值處理方法包括刪除異常值、替換異常值、截尾等,可提高數(shù)據(jù)準(zhǔn)確性及可靠性。

降雨量數(shù)據(jù)正常范圍應(yīng)在[0,100],爬取的數(shù)據(jù)中出現(xiàn)了明顯超過正常范圍的數(shù)值,如178、-1等,基于異常值刪除和異常值替換對(duì)其進(jìn)行處理。異常值刪除:將降雨量字段中大于100的值刪除。異常值替換:對(duì)降雨量字段中小于0的值使用0替換。

2)重復(fù)值。重復(fù)記錄與數(shù)據(jù)采集及傳輸?shù)仍蚩赡軐?dǎo)致數(shù)據(jù)集中存在部分重復(fù)值,即同一時(shí)間和地點(diǎn)下有多個(gè)數(shù)據(jù)記錄,這樣的數(shù)據(jù)對(duì)分析結(jié)果有一定的影響,需進(jìn)行數(shù)據(jù)預(yù)處理。采集到的數(shù)據(jù)集氣壓存在重復(fù)部分,針對(duì)重復(fù)的數(shù)據(jù)采用合并相加取平均值的方式處理。

3)缺失值。由于種種原因,爬取的氣象數(shù)據(jù)可能存在部分缺失的情況,在數(shù)據(jù)清洗與預(yù)處理過程中需對(duì)這些缺失值進(jìn)行處理,以保證數(shù)據(jù)的完整性及準(zhǔn)確性。爬取的數(shù)據(jù)中溫度字段存在缺失值,采用相鄰兩項(xiàng)相加取平均值的方式進(jìn)行填充,具體計(jì)算方法如公式1所示。

Tn=(Tn-1+Tn+1)/2

(1)

3 實(shí)時(shí)氣象數(shù)據(jù)分析

在實(shí)時(shí)氣候分析模塊需通過Spark SQL分析原始數(shù)據(jù)集,得到氣象基本信息表和天氣狀況表,用以繪制一天內(nèi)溫度變化、一天內(nèi)相對(duì)濕度變化、一天內(nèi)空氣質(zhì)量變化、一天風(fēng)級(jí)變化、未來14 d高溫低溫變化、未來14 d風(fēng)級(jí)變化、未來14 d氣候分布等圖表。

對(duì)于氣象基本信息表,需獲取原始數(shù)據(jù)集,將其轉(zhuǎn)換為DataFrame格式,使用Spark SQL從原始數(shù)據(jù)集中提取整點(diǎn)時(shí)間、溫度、風(fēng)向、降雨量、濕度、空氣質(zhì)量及城市字段。整點(diǎn)時(shí)間:使用hour函數(shù)從日期字段中提取整點(diǎn)時(shí)間,例如日期字段為2022-04-13 14∶23∶45,則整點(diǎn)時(shí)間為14。風(fēng)向:使用udf函數(shù)將原始數(shù)據(jù)集中的風(fēng)向字段轉(zhuǎn)換為英文縮寫形式,北風(fēng)轉(zhuǎn)換為N,東北風(fēng)轉(zhuǎn)換為NE。空氣質(zhì)量:使用case when函數(shù)將空氣質(zhì)量字段劃分為優(yōu)、良和一般三個(gè)等級(jí)。溫度、降雨量、濕度、城市字段直接使用原始數(shù)據(jù)集中的數(shù)據(jù)。

對(duì)于天氣狀況表,需從原始數(shù)據(jù)集中提取日期、氣壓、溫度、降水量、風(fēng)向及城市字段。天氣狀況:根據(jù)原始數(shù)據(jù)集中的氣壓、濕度及降水量字段,使用when函數(shù)和otherwise函數(shù)來判斷天氣狀況。根據(jù)降雨量字段判斷:降水量大于100,為大雨;降雨量大于50小于100,為中雨;降雨量大于0小于50,為小雨;降雨量為0,則根據(jù)濕度判斷。濕度大于60,為陰;小于60則根據(jù)氣壓判斷。氣壓小于1000,為多云,否則為晴天。最高/低溫度:選取原始數(shù)據(jù)集中溫度字段和日期字段,將日期字段轉(zhuǎn)化為日期類型yyyy/MM/dd,按照日期和城市分組,使用man/min函數(shù)計(jì)算每個(gè)分組中的最高/低溫度。風(fēng)向1和風(fēng)向2:將原始字段的風(fēng)向值除以180°得到一個(gè)值,將這個(gè)值拆分成兩個(gè)數(shù),一個(gè)為sin值,一個(gè)為cos值,用arctan 2函數(shù)將這兩個(gè)數(shù)轉(zhuǎn)化為角度,轉(zhuǎn)化后的sin值和cos值分別對(duì)應(yīng)風(fēng)向1和風(fēng)向2。城市:使用原始數(shù)據(jù)集中的城市字段。

4 結(jié)構(gòu)化數(shù)據(jù)存儲(chǔ)

基于氣象數(shù)據(jù)分析結(jié)果設(shè)計(jì)了兩張結(jié)構(gòu)化數(shù)據(jù)庫(kù)表,即氣象基本信息表和天氣狀況表,根據(jù)這兩張表中的數(shù)據(jù)繪制可視化分析圖表[8]。對(duì)兩張表的邏輯結(jié)構(gòu)和物理結(jié)構(gòu)分析如下。

氣象基本信息表主要包括ID、溫度、空氣質(zhì)量、風(fēng)向、降雨量、城市、溫度、整點(diǎn)時(shí)間等屬性。邏輯結(jié)構(gòu)設(shè)計(jì)如下。

物理結(jié)構(gòu)設(shè)計(jì)如表1所示。

表1 氣象基本信息表物理結(jié)構(gòu)

天氣狀況表主要包括ID、日期、天氣狀況、最低溫度、最高溫度、風(fēng)向1、風(fēng)向2、風(fēng)速、城市等屬性,邏輯結(jié)構(gòu)設(shè)計(jì)如下。

物理結(jié)構(gòu)設(shè)計(jì)如表2所示。

表2 天氣狀況表物理結(jié)構(gòu)

5 數(shù)據(jù)可視化展示

數(shù)據(jù)可視化部分主要對(duì)當(dāng)天的濕度、溫度、空氣質(zhì)量、風(fēng)級(jí)等做實(shí)時(shí)展示,并對(duì)未來7 d和14 d的氣候分布、溫度變化等做出預(yù)測(cè)。

1)相對(duì)濕度曲線。根據(jù)天氣基本信息表中的整點(diǎn)時(shí)間和相對(duì)濕度值計(jì)算平均相對(duì)濕度、最高相對(duì)濕度、最低相對(duì)濕度的值及其所在的小時(shí)數(shù),將一天內(nèi)24h分布作為X軸,獲得的相對(duì)濕度作為Y軸,繪制一天內(nèi)的相對(duì)濕度曲線。

2)溫度變化曲線。根據(jù)天氣基本信息表中的整點(diǎn)時(shí)間和溫度數(shù)據(jù),將整點(diǎn)時(shí)間作為X軸,溫度作為Y軸,繪制一天內(nèi)的溫度變化曲線。

3)空氣質(zhì)量變化。根據(jù)天氣基本信息表中的整點(diǎn)時(shí)間和空氣質(zhì)量數(shù)據(jù),將整點(diǎn)時(shí)間作為X軸,平均空氣質(zhì)量作為Y軸,繪制一天內(nèi)的空氣質(zhì)量變化柱狀圖。

4)風(fēng)級(jí)圖。將天氣狀況數(shù)據(jù)表中的風(fēng)向1和風(fēng)向2轉(zhuǎn)換為角度值,以45°為間隔劃分不同的角度區(qū)間,計(jì)算每個(gè)區(qū)間內(nèi)的風(fēng)速平均值,得到一個(gè)包含8個(gè)值的列表,使用極區(qū)圖繪制風(fēng)向與風(fēng)速的分布情況,如圖7所示。

圖7 風(fēng)級(jí)圖Fig.7 Wind scale chart

5)未來14 d氣候分布。在天氣狀況數(shù)據(jù)表中獲取未來14 d的天氣數(shù)據(jù),循環(huán)對(duì)每種天氣出現(xiàn)的次數(shù)進(jìn)行計(jì)數(shù),并將結(jié)果保存在字典中,將字典的鍵值作為標(biāo)簽,字典的值用于表示餅圖中每個(gè)扇區(qū)的大小,未來14 d氣候分布繪制如圖8所示。

圖8 未來14天氣候分布Fig.8 Climate distribution in the next 14 days

6)最高和最低溫度變化曲線。將天氣狀況表中的日期、最低氣溫和最高氣溫?cái)?shù)據(jù)提取出來,計(jì)算每天的最高/最低溫度及最高/最低溫度的平均值。利用溫度數(shù)據(jù)繪制高溫度曲線、低溫度曲線、平均溫度虛線,如圖9所示。

圖9 未來14 d高低溫曲線Fig.9 Temperature curve in future 14 days

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

設(shè)計(jì)了一種基于Spark的實(shí)時(shí)氣象數(shù)據(jù)分析流程,可實(shí)時(shí)獲取并處理數(shù)據(jù),并將數(shù)據(jù)轉(zhuǎn)化為直觀易懂的可視化展示,使人們更好地理解和分析天氣情況。對(duì)于實(shí)時(shí)數(shù)據(jù)的處理,使用Kafka消息系統(tǒng),Kafka的分布式架構(gòu)和高吞吐量特性確保了可靠、高效的氣象數(shù)據(jù)傳輸,保障了數(shù)據(jù)的準(zhǔn)確性,能夠處理并存儲(chǔ)大規(guī)模的氣象數(shù)據(jù)。此設(shè)計(jì)流程與方法可快速處理分析大規(guī)模的氣象數(shù)據(jù),提供準(zhǔn)確的分析結(jié)果和強(qiáng)大的可視化功能。應(yīng)用此設(shè)計(jì)系統(tǒng)可更好地應(yīng)對(duì)不斷增加的氣象數(shù)據(jù),提供準(zhǔn)確的氣象信息及預(yù)測(cè),為社會(huì)各個(gè)領(lǐng)域提供有價(jià)值的支持和決策依據(jù)。