基于Spark技術的氣象數據分析

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

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

0 引言

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

系統主要流程包括實時氣象數據采集、數據清洗與預處理、數據處理與分析、結構化數據存儲、數據可視化等部分,利用網絡爬蟲技術采集中央氣象臺的氣象數據,將數據發送到Kafka消息隊列[4]中。使用Spark技術快速高效分析數據,將分析得到的結果存儲在MySQL中,基于Spring Boot框架搭建web系統管理數據并運用Echarts進行數據分析及可視化展示,具體過程如圖1所示。

圖1 數據分析處理流程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 一天內相對濕度曲線值Fig.4 Relative humidity curve on one day

圖5 一天內溫度曲線值Fig.5 Temperature curve on one day

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

采集陜西省各城市氣象數據,數據主要來源于中央氣象臺官方網站。數據采集流程具體為:獲取陜西省及省內各城市的編碼,進行記錄,分析各城市24 h整點天氣頁面,分析頁面結構、頁面源代碼等,基于爬蟲技術爬取各城市24 h的整點天氣數據,將數據打包成JSON格式的消息發送到Kafka隊列中?；谂老x技術快速獲取實時數據,設置定時器功能,設定每60 min執行一次爬蟲程序,以獲取準確的實時數據。爬蟲程序將爬取到的數據通過BeautifulSoup庫來解析數據,轉換為python對象[5]。提取的數據主要包括氣溫、城市、日期、風速、風向、濕度、降水量、氣壓等,提取到的數據最終被轉換為JSON格式的字符串,以便在Kafka隊列中進行傳輸。

1 Kafka數據傳輸

經爬蟲采集到的氣象數據被封裝成JSON字符串后發送到Kafka中名為“weather_data”的Topic[6]。Kafka以分布式方式存儲數據,數據被復制到多個Broker上,Broker接收并持久化發布到Topic的數據,確保數據的高可用性及容錯性。客戶端在獲取氣象數據時,使用Kafka的Consumer客戶端訂閱“weather_data”這個Topic,Kafka將數據傳輸給訂閱該Topic的客戶端,客戶端可以按照自己的速率處理數據。整個過程中,Kafka的分布式架構和高吞吐量特性確保了可靠、高效的氣象數據傳輸,保證數據能夠準時送達消費者,在數據量增大時也能夠處理并存儲大規模的氣象數據[7]。

2 數據清洗與預處理

數據清洗和預處理是在對數據進行審查及校驗過程中發現并糾正數據文件中可識別的錯誤,按照一定的規則糾正錯誤或沖突數據。通用的數據清洗與預處理主要包括處理異常數據、重復數值及缺失值等。

1)異常數值。采集的數據往往存在一些異常值,這些異常值可能由于數據采集錯誤、傳輸錯誤等原因導致,會嚴重影響數據分析結果,因此需要進行異常值處理。常用的異常值處理方法包括刪除異常值、替換異常值、截尾等,可提高數據準確性及可靠性。

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

2)重復值。重復記錄與數據采集及傳輸等原因可能導致數據集中存在部分重復值,即同一時間和地點下有多個數據記錄,這樣的數據對分析結果有一定的影響,需進行數據預處理。采集到的數據集氣壓存在重復部分,針對重復的數據采用合并相加取平均值的方式處理。

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

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

(1)

3 實時氣象數據分析

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

對于氣象基本信息表,需獲取原始數據集,將其轉換為DataFrame格式,使用Spark SQL從原始數據集中提取整點時間、溫度、風向、降雨量、濕度、空氣質量及城市字段。整點時間:使用hour函數從日期字段中提取整點時間,例如日期字段為2022-04-13 14∶23∶45,則整點時間為14。風向:使用udf函數將原始數據集中的風向字段轉換為英文縮寫形式,北風轉換為N,東北風轉換為NE?？諝赓|量:使用case when函數將空氣質量字段劃分為優、良和一般三個等級。溫度、降雨量、濕度、城市字段直接使用原始數據集中的數據。

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

4 結構化數據存儲

基于氣象數據分析結果設計了兩張結構化數據庫表,即氣象基本信息表和天氣狀況表,根據這兩張表中的數據繪制可視化分析圖表[8]。對兩張表的邏輯結構和物理結構分析如下。

氣象基本信息表主要包括ID、溫度、空氣質量、風向、降雨量、城市、溫度、整點時間等屬性。邏輯結構設計如下。

物理結構設計如表1所示。

表1 氣象基本信息表物理結構

天氣狀況表主要包括ID、日期、天氣狀況、最低溫度、最高溫度、風向1、風向2、風速、城市等屬性,邏輯結構設計如下。

物理結構設計如表2所示。

表2 天氣狀況表物理結構

5 數據可視化展示

數據可視化部分主要對當天的濕度、溫度、空氣質量、風級等做實時展示,并對未來7 d和14 d的氣候分布、溫度變化等做出預測。

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

2)溫度變化曲線。根據天氣基本信息表中的整點時間和溫度數據,將整點時間作為X軸,溫度作為Y軸,繪制一天內的溫度變化曲線。

3)空氣質量變化。根據天氣基本信息表中的整點時間和空氣質量數據,將整點時間作為X軸,平均空氣質量作為Y軸,繪制一天內的空氣質量變化柱狀圖。

4)風級圖。將天氣狀況數據表中的風向1和風向2轉換為角度值,以45°為間隔劃分不同的角度區間,計算每個區間內的風速平均值,得到一個包含8個值的列表,使用極區圖繪制風向與風速的分布情況,如圖7所示。

圖7 風級圖Fig.7 Wind scale chart

5)未來14 d氣候分布。在天氣狀況數據表中獲取未來14 d的天氣數據,循環對每種天氣出現的次數進行計數,并將結果保存在字典中,將字典的鍵值作為標簽,字典的值用于表示餅圖中每個扇區的大小,未來14 d氣候分布繪制如圖8所示。

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

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

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

6 結束語

設計了一種基于Spark的實時氣象數據分析流程,可實時獲取并處理數據,并將數據轉化為直觀易懂的可視化展示,使人們更好地理解和分析天氣情況。對于實時數據的處理,使用Kafka消息系統,Kafka的分布式架構和高吞吐量特性確保了可靠、高效的氣象數據傳輸,保障了數據的準確性,能夠處理并存儲大規模的氣象數據。此設計流程與方法可快速處理分析大規模的氣象數據,提供準確的分析結果和強大的可視化功能。應用此設計系統可更好地應對不斷增加的氣象數據,提供準確的氣象信息及預測,為社會各個領域提供有價值的支持和決策依據。