1. 從一份氣象數(shù)據(jù)說起為什么Spark是處理它的不二之選幾年前我接手過一個(gè)項(xiàng)目需要分析全國上千個(gè)氣象站點(diǎn)過去十年的分鐘級(jí)觀測(cè)數(shù)據(jù)目標(biāo)是找出特定區(qū)域的極端天氣模式。數(shù)據(jù)量不算天文數(shù)字但也達(dá)到了TB級(jí)別。最初嘗試用傳統(tǒng)的關(guān)系型數(shù)據(jù)庫和單機(jī)腳本結(jié)果一個(gè)簡單的關(guān)聯(lián)查詢就能讓系統(tǒng)卡上幾個(gè)小時(shí)更別提復(fù)雜的時(shí)空序列分析了。直到我們把計(jì)算引擎切換到Spark整個(gè)局面才豁然開朗。那個(gè)項(xiàng)目讓我深刻體會(huì)到面對(duì)氣象數(shù)據(jù)這種典型的時(shí)空大數(shù)據(jù)選對(duì)工具是多么關(guān)鍵。氣象數(shù)據(jù)分析聽起來像是個(gè)科研課題但實(shí)際上它的應(yīng)用場景早已滲透到我們生活的方方面面。從你手機(jī)上的天氣預(yù)報(bào)App到電網(wǎng)的負(fù)荷預(yù)測(cè)、農(nóng)業(yè)的災(zāi)害預(yù)警、航空公司的航線規(guī)劃背后都離不開對(duì)海量氣象數(shù)據(jù)的實(shí)時(shí)或離線處理。這些數(shù)據(jù)通常具有幾個(gè)鮮明的“大數(shù)據(jù)”特征體量大全球觀測(cè)網(wǎng)絡(luò)、衛(wèi)星遙感每天都在產(chǎn)生PB級(jí)的數(shù)據(jù)、速度快數(shù)據(jù)流持續(xù)不斷、多樣性高包括結(jié)構(gòu)化觀測(cè)記錄、半結(jié)構(gòu)化報(bào)文、非結(jié)構(gòu)化衛(wèi)星云圖等。處理這類數(shù)據(jù)傳統(tǒng)的單機(jī)工具或小型數(shù)據(jù)庫往往力不從心。而Spark正是為應(yīng)對(duì)這種挑戰(zhàn)而生的。它不是一個(gè)單一的軟件而是一個(gè)統(tǒng)一的、內(nèi)存優(yōu)先的分布式計(jì)算框架。它的核心優(yōu)勢(shì)在于能將一個(gè)龐大的計(jì)算任務(wù)自動(dòng)分解成無數(shù)個(gè)小任務(wù)分發(fā)到成百上千臺(tái)普通的服務(wù)器上并行執(zhí)行最后再把結(jié)果匯總起來。這種“分而治之”的思想完美契合了氣象數(shù)據(jù)“量大但可分割”的特性。比如要計(jì)算每個(gè)省份的年平均氣溫Spark可以輕松地將數(shù)據(jù)按省份分區(qū)在不同機(jī)器上同時(shí)計(jì)算效率呈線性提升。更重要的是Spark提供了一套高層API如Spark SQL、DataFrame讓數(shù)據(jù)分析師可以用接近SQL或Python Pandas的方式去操作分布式數(shù)據(jù)而不必深究底層復(fù)雜的分布式系統(tǒng)細(xì)節(jié)。這對(duì)于氣象、環(huán)保等領(lǐng)域的業(yè)務(wù)專家來說極大地降低了大數(shù)據(jù)處理的門檻。你可以專注于數(shù)據(jù)本身的規(guī)律和業(yè)務(wù)邏輯而不是糾結(jié)于如何調(diào)優(yōu)一個(gè)MapReduce作業(yè)。所以當(dāng)你手頭有一批氣象數(shù)據(jù)想要挖掘其價(jià)值時(shí)基于Spark構(gòu)建分析流程幾乎是一個(gè)自然而然的現(xiàn)代選擇。它不僅解決了算力瓶頸更提供了一套高效、易用且生態(tài)豐富的工具鏈。接下來我將結(jié)合一個(gè)從數(shù)據(jù)準(zhǔn)備到分析可視化的完整案例拆解其中的核心技術(shù)點(diǎn)、實(shí)操步驟以及那些容易踩坑的細(xì)節(jié)。2. 實(shí)戰(zhàn)環(huán)境搭建從零部署一個(gè)可用的Spark集群工欲善其事必先利其器。在開始寫分析代碼之前一個(gè)穩(wěn)定、高效的Spark運(yùn)行環(huán)境是基礎(chǔ)。對(duì)于個(gè)人學(xué)習(xí)或中小型項(xiàng)目我強(qiáng)烈推薦使用Local模式或Standalone集群模式起步完全沒必要一開始就上復(fù)雜的YARN或Kubernetes。2.1 基礎(chǔ)環(huán)境準(zhǔn)備與Spark安裝首先我們需要一個(gè)Linux環(huán)境Ubuntu Server是一個(gè)穩(wěn)妥的選擇。假設(shè)你已經(jīng)在虛擬機(jī)或云服務(wù)器上安裝好了Ubuntu 22.04 LTS。第一步是安裝JavaSpark運(yùn)行在JVM之上。OpenJDK 8或11是經(jīng)過廣泛驗(yàn)證的穩(wěn)定版本。# 更新包列表 sudo apt update # 安裝OpenJDK 11 sudo apt install openjdk-11-jdk-headless -y # 驗(yàn)證安裝 java -version接下來下載Spark。訪問Apache Spark官網(wǎng)的 下載頁面 。這里有個(gè)關(guān)鍵選擇Pre-built for Apache Hadoop版本。即使你不使用HDFS也建議選擇這個(gè)版本因?yàn)樗伺cHadoop生態(tài)系統(tǒng)交互所需的庫。我們選擇最新的穩(wěn)定版例如Spark 3.5.x包類型選“Pre-built for Apache Hadoop 3.3 and later”。# 進(jìn)入常用安裝目錄例如/opt cd /opt # 使用wget下載請(qǐng)?zhí)鎿Q為官網(wǎng)最新的實(shí)際鏈接 sudo wget https://dlcdn.apache.org/spark/spark-3.5.0/spark-3.5.0-bin-hadoop3.tgz # 解壓 sudo tar -xzf spark-3.5.0-bin-hadoop3.tgz # 創(chuàng)建一個(gè)軟鏈接方便后續(xù)版本升級(jí) sudo ln -s spark-3.5.0-bin-hadoop3 spark然后需要設(shè)置環(huán)境變量將Spark的bin目錄加入PATH并設(shè)置SPARK_HOME。# 編輯當(dāng)前用戶的bash配置文件 nano ~/.bashrc在文件末尾添加export SPARK_HOME/opt/spark export PATH$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin保存退出后執(zhí)行source ~/.bashrc使配置生效?，F(xiàn)在你可以通過運(yùn)行spark-shellScala交互式環(huán)境或pysparkPython交互式環(huán)境來快速驗(yàn)證安裝是否成功。如果看到一個(gè)帶著Spark Logo的交互式命令行并且沒有報(bào)錯(cuò)說明Spark本地模式已經(jīng)可以運(yùn)行了。2.2 集群模式配置要點(diǎn)與資源規(guī)劃Local模式適合測(cè)試和調(diào)試但處理真實(shí)數(shù)據(jù)時(shí)我們需要利用多臺(tái)機(jī)器的資源也就是集群模式。Spark Standalone模式是Spark自帶的輕量級(jí)集群管理器配置簡單足以應(yīng)對(duì)很多生產(chǎn)場景。假設(shè)我們有三臺(tái)機(jī)器一臺(tái)主節(jié)點(diǎn)master兩臺(tái)工作節(jié)點(diǎn)worker1, worker2。首先在所有節(jié)點(diǎn)上重復(fù)上述Spark安裝步驟。在主節(jié)點(diǎn)上配置進(jìn)入$SPARK_HOME/conf目錄配置核心文件。配置spark-env.sh復(fù)制模板文件并編輯。cd /opt/spark/conf cp spark-env.sh.template spark-env.sh nano spark-env.sh添加以下內(nèi)容根據(jù)你的機(jī)器配置調(diào)整# 指定Master節(jié)點(diǎn)的IP或主機(jī)名 export SPARK_MASTER_HOSTyour_master_ip # 指定Master Web UI端口 export SPARK_MASTER_WEBUI_PORT8080 # 指定每個(gè)Worker節(jié)點(diǎn)能使用的最大CPU核心數(shù) export SPARK_WORKER_CORES4 # 指定每個(gè)Worker節(jié)點(diǎn)能使用的最大內(nèi)存注意單位是MB且要預(yù)留一部分給系統(tǒng)和其他進(jìn)程 export SPARK_WORKER_MEMORY8g # 指定Spark日志目錄 export SPARK_LOG_DIR/opt/spark/logs配置slaves文件指定所有工作節(jié)點(diǎn)。cp slaves.template slaves nano slaves在文件中添加工作節(jié)點(diǎn)的主機(jī)名或IP每行一個(gè)worker1 worker2在工作節(jié)點(diǎn)上配置工作節(jié)點(diǎn)只需要配置spark-env.sh中的資源參數(shù)SPARK_WORKER_CORES和SPARK_WORKER_MEMORY確保其值小于或等于該節(jié)點(diǎn)的實(shí)際物理資源。配置SSH免密登錄這是Standalone集群啟動(dòng)的關(guān)鍵。主節(jié)點(diǎn)需要能通過SSH無密碼登錄到所有工作節(jié)點(diǎn)包括自己。在主節(jié)點(diǎn)上執(zhí)行# 生成密鑰對(duì)如果已有可跳過 ssh-keygen -t rsa # 將公鑰復(fù)制到所有節(jié)點(diǎn)包括本機(jī) ssh-copy-id your_master_ip ssh-copy-id worker1 ssh-copy-id worker2完成后測(cè)試從主節(jié)點(diǎn)ssh worker1能否直接登錄。啟動(dòng)與驗(yàn)證集群在主節(jié)點(diǎn)上進(jìn)入$SPARK_HOME/sbin目錄。# 啟動(dòng)Master和所有Slaves ./start-all.sh啟動(dòng)后在主節(jié)點(diǎn)上運(yùn)行jps命令應(yīng)該能看到Master進(jìn)程在工作節(jié)點(diǎn)上運(yùn)行jps應(yīng)該能看到Worker進(jìn)程。訪問主節(jié)點(diǎn)的8080端口如http://your_master_ip:8080你將看到Spark Standalone集群的Web UI上面清晰地展示了集群的資源狀態(tài)和運(yùn)行的應(yīng)用程序。注意關(guān)于資源規(guī)劃SPARK_WORKER_MEMORY的設(shè)置是個(gè)技術(shù)活。如果你給Worker分配了8g內(nèi)存Spark內(nèi)部會(huì)將其分為兩部分一部分用于執(zhí)行內(nèi)存Execution Memory用于shuffle、join、aggregation等計(jì)算一部分用于存儲(chǔ)內(nèi)存Storage Memory用于緩存RDD/DataFrame。默認(rèn)比例是執(zhí)行內(nèi)存占0.6存儲(chǔ)內(nèi)存占0.4。如果任務(wù)需要大量緩存可以適當(dāng)調(diào)高spark.memory.storageFraction如果任務(wù)shuffle很重則可以調(diào)低。一個(gè)常見的坑是分配的內(nèi)存超過物理內(nèi)存導(dǎo)致OOMOut Of Memory錯(cuò)誤所以務(wù)必預(yù)留至少1-2G給操作系統(tǒng)和其他服務(wù)。2.3 開發(fā)工具鏈與依賴管理對(duì)于氣象數(shù)據(jù)分析Python因其豐富的數(shù)據(jù)科學(xué)生態(tài)Pandas, NumPy, Matplotlib, Scikit-learn而成為主流選擇。Spark通過PySpark提供了完整的Python API。我推薦使用Jupyter Notebook或JupyterLab作為交互式開發(fā)環(huán)境它非常適合數(shù)據(jù)探索和可視化。你可以通過Anaconda或Miniconda來管理Python環(huán)境。# 安裝Miniconda wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh bash Miniconda3-latest-Linux-x86_64.sh # 創(chuàng)建一個(gè)專門的Spark環(huán)境 conda create -n spark-env python3.9 conda activate spark-env # 安裝常用庫 pip install jupyterlab pyspark pandas numpy matplotlib seaborn為了讓Jupyter能使用我們安裝的Spark需要設(shè)置一些環(huán)境變量。在你的Jupyter啟動(dòng)腳本或~/.bashrc中確保設(shè)置了PYSPARK_PYTHON指向conda環(huán)境中的Python解釋器。export PYSPARK_PYTHON/path/to/your/miniconda3/envs/spark-env/bin/python啟動(dòng)JupyterLab后你就可以在Notebook中創(chuàng)建SparkSession了這是所有Spark功能的入口點(diǎn)。from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(WeatherDataAnalysis) \ .master(spark://your_master_ip:7077) \ # 連接到Standalone集群 .config(spark.executor.memory, 4g) \ # 每個(gè)執(zhí)行器內(nèi)存 .config(spark.driver.memory, 2g) \ # 驅(qū)動(dòng)器內(nèi)存本地客戶端 .getOrCreate()實(shí)操心得在團(tuán)隊(duì)協(xié)作中依賴管理是個(gè)大問題。PySpark作業(yè)可能會(huì)依賴第三方Python包如scikit-learn。有幾種解決方案1) 在所有集群節(jié)點(diǎn)上手動(dòng)安裝相同版本的包繁瑣且易出錯(cuò)2) 使用Spark的--py-files參數(shù)提交壓縮的依賴包3) 使用conda-pack將整個(gè)conda環(huán)境打包通過spark.submit.pyFiles分發(fā)。對(duì)于生產(chǎn)環(huán)境我傾向于第三種它能最大程度保證環(huán)境一致性。此外對(duì)于Scala/Java項(xiàng)目則需要使用Maven或SBT來管理JAR包依賴并通過--jars參數(shù)提交。3. 氣象數(shù)據(jù)的獲取、理解與預(yù)處理有了環(huán)境下一步就是處理數(shù)據(jù)本身。氣象數(shù)據(jù)來源多樣格式不一質(zhì)量參差。這一步做得好后續(xù)分析事半功倍做得不好則可能“垃圾進(jìn)垃圾出”。3.1 數(shù)據(jù)源概覽與獲取途徑氣象數(shù)據(jù)主要分為以下幾類地面觀測(cè)數(shù)據(jù)來自氣象站記錄溫度、氣壓、濕度、降水量、風(fēng)速風(fēng)向等。格式多為CSV、TXT或特定的二進(jìn)制格式如BUFR。國內(nèi)可以從國家氣象信息中心等機(jī)構(gòu)獲取國際上則有NOAA的GSOD、NCDC等公開數(shù)據(jù)集。高空探測(cè)數(shù)據(jù)探空儀數(shù)據(jù)提供不同氣壓層的氣象要素。格式通常為特定編碼如TEMP, PILOT。雷達(dá)數(shù)據(jù)基數(shù)據(jù)如NEXRAD Level II體積龐大處理復(fù)雜通常用于專業(yè)研究。衛(wèi)星數(shù)據(jù)如風(fēng)云、GOES、MODIS等衛(wèi)星的遙感產(chǎn)品格式多為HDF或NetCDF包含多光譜通道信息。數(shù)值預(yù)報(bào)模式輸出如WRF、ECMWF等模式生成的格點(diǎn)數(shù)據(jù)格式多為GRIB或NetCDF。對(duì)于學(xué)習(xí)和原型開發(fā)我推薦從公開的、結(jié)構(gòu)化的地面觀測(cè)數(shù)據(jù)開始。例如我們可以使用NOAA的全球歷史氣候網(wǎng)絡(luò)日數(shù)據(jù)GHCN-Daily。它包含了全球數(shù)萬個(gè)站點(diǎn)的日值數(shù)據(jù)可以通過FTP或API下載。假設(shè)我們下載了一個(gè)CSV文件ghcnd-stations.txt站點(diǎn)元數(shù)據(jù)和一批以.dly為后綴的日值數(shù)據(jù)文件。原始數(shù)據(jù)往往不是“整潔”的每個(gè).dly文件包含了某個(gè)站點(diǎn)多個(gè)氣象要素變量的多年記錄是一種“寬表”格式需要解析。3.2 使用Spark DataFrame進(jìn)行數(shù)據(jù)加載與解析Spark支持多種數(shù)據(jù)源。對(duì)于CSV、JSON等結(jié)構(gòu)化/半結(jié)構(gòu)化數(shù)據(jù)spark.readAPI是首選。但對(duì)于自定義格式的.dly文件我們需要先進(jìn)行解析。首先查看數(shù)據(jù)格式。GHCN-Daily的日值數(shù)據(jù)文件每行固定長度包含了站點(diǎn)ID、年月日、要素代碼以及31天的值每個(gè)值占固定列寬。我們可以編寫一個(gè)解析函數(shù)用Spark的RDDAPI或map函數(shù)來處理。from pyspark.sql import Row from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, DateType # 定義數(shù)據(jù)模式Schema schema StructType([ StructField(station_id, StringType(), False), StructField(date, DateType(), False), StructField(element, StringType(), False), # 要素代碼如TMAX, TMIN, PRCP StructField(value, DoubleType(), True), # 觀測(cè)值 StructField(m_flag, StringType(), True), # 測(cè)量標(biāo)志 StructField(q_flag, StringType(), True), # 質(zhì)量標(biāo)志 StructField(s_flag, StringType(), True), # 來源標(biāo)志 ]) def parse_ghcn_line(line): 解析GHCN日數(shù)據(jù)的一行記錄 station_id line[0:11] year int(line[11:15]) month int(line[15:17]) element line[17:21] values [] # 解析31天的數(shù)據(jù) for i in range(31): start 21 i * 8 end start 5 # 值占5位可能為-9999缺失值 str_val line[start:end].strip() value float(str_val) / 10.0 if str_val ! -9999 else None # 注意單位轉(zhuǎn)換如溫度是0.1度 # 標(biāo)志位1位 m_flag line[start5:start6] if len(line) start5 else q_flag line[start6:start7] if len(line) start6 else s_flag line[start7:start8] if len(line) start7 else if value is not None: # 只生成有效日期的記錄 day i 1 # 注意處理閏年、月份天數(shù)這里簡化處理 try: from datetime import date record_date date(year, month, day) values.append(Row(station_idstation_id, daterecord_date, elementelement, valuevalue, m_flagm_flag, q_flagq_flag, s_flags_flag)) except ValueError: # 無效日期如2月30日跳過 pass return values # 加載原始文本文件 raw_rdd spark.sparkContext.textFile(hdfs://path/to/your/*.dly) # 或本地路徑 file:// # 應(yīng)用解析函數(shù)并扁平化 parsed_rdd raw_rdd.flatMap(parse_ghcn_line) # 轉(zhuǎn)換為DataFrame weather_df spark.createDataFrame(parsed_rdd, schemaschema) weather_df.cache() # 緩存起來因?yàn)楹罄m(xù)會(huì)多次使用 weather_df.show(5)這段代碼展示了如何將非標(biāo)準(zhǔn)格式的數(shù)據(jù)轉(zhuǎn)化為Spark DataFrame。flatMap操作符非常適合這種“一行輸入多行輸出”的解析邏輯。定義明確的schema不僅能提高效率還能在后續(xù)的SQL查詢中享受Catalyst優(yōu)化器的性能紅利。3.3 數(shù)據(jù)質(zhì)量清洗與特征工程關(guān)鍵步驟原始數(shù)據(jù)必然存在缺失、異常和錯(cuò)誤。清洗是數(shù)據(jù)分析的基石。1. 處理缺失值Spark DataFrame提供了靈活的缺失值處理方式。# 查看缺失情況 from pyspark.sql.functions import col, count, when, isnan, isnull weather_df.select([count(when(isnull(c) | isnan(c), c)).alias(c) for c in weather_df.columns]).show() # 策略1刪除缺失值過多的記錄謹(jǐn)慎使用可能引入偏差 # 例如刪除value為空的記錄 cleaned_df weather_df.filter(col(value).isNotNull()) # 策略2填充缺失值 # 對(duì)于溫度可以用前后天的平均值填充需要窗口函數(shù) from pyspark.sql.window import Window from pyspark.sql.functions import avg, lag, lead window_spec Window.partitionBy(station_id, element).orderBy(date) # 計(jì)算前后兩天的平均值 df_with_avg cleaned_df.withColumn(prev_val, lag(value, 1).over(window_spec)) \ .withColumn(next_val, lead(value, 1).over(window_spec)) df_filled df_with_avg.withColumn(value_filled, when(col(value).isNotNull(), col(value)) .otherwise((col(prev_val) col(next_val)) / 2))2. 處理異常值基于物理常識(shí)進(jìn)行過濾。例如地表溫度通常在一定范圍內(nèi)。# 過濾掉明顯異常的溫度值單位攝氏度 valid_temp_df df_filled.filter( ~((col(element) TMAX) ((col(value_filled) 60) | (col(value_filled) -90))) ~((col(element) TMIN) ((col(value_filled) 50) | (col(value_filled) -90))) )3. 數(shù)據(jù)轉(zhuǎn)換與特征工程為了便于分析我們常常需要轉(zhuǎn)換數(shù)據(jù)形態(tài)或創(chuàng)建新特征。數(shù)據(jù)透視Pivot將“長格式”數(shù)據(jù)一行一個(gè)要素轉(zhuǎn)為“寬格式”一行包含所有要素。# 將不同要素TMAX, TMIN, PRCP變成不同的列 wide_df valid_temp_df.groupBy(station_id, date).pivot(element).avg(value_filled) wide_df wide_df.withColumnRenamed(TMAX, tmax) \ .withColumnRenamed(TMIN, tmin) \ .withColumnRenamed(PRCP, prcp) wide_df.show(5)創(chuàng)建衍生特征例如計(jì)算日平均溫度、溫度日較差、累計(jì)降水量等。from pyspark.sql.functions import coalesce wide_df wide_df.withColumn(tavg, (col(tmax) col(tmin)) / 2.0) \ .withColumn(trange, col(tmax) - col(tmin)) # 計(jì)算每個(gè)站點(diǎn)每月的累計(jì)降水量假設(shè)prcp單位是mm monthly_prcp_df wide_df.filter(col(prcp).isNotNull()) \ .groupBy(station_id, year(date).alias(year), month(date).alias(month)) \ .agg(sum(prcp).alias(monthly_prcp))踩坑實(shí)錄數(shù)據(jù)透視操作pivot在要素類別即element列的不同值非常多時(shí)會(huì)導(dǎo)致生成的列數(shù)爆炸嚴(yán)重消耗內(nèi)存和性能。在實(shí)際操作中一定要先檢查唯一要素的數(shù)量df.select(element).distinct().count()。如果數(shù)量過大比如超過1000就需要考慮其他策略比如分批次處理或者保持長格式使用filter來分別處理不同要素。另一個(gè)常見問題是時(shí)區(qū)。原始數(shù)據(jù)中的日期時(shí)間字段可能沒有時(shí)區(qū)信息或者使用的是UTC。在涉及跨時(shí)區(qū)站點(diǎn)的分析時(shí)必須統(tǒng)一時(shí)區(qū)處理否則會(huì)導(dǎo)致日界劃分錯(cuò)誤。我通常的做法是在數(shù)據(jù)加載后立即使用from_utc_timestamp函數(shù)將所有時(shí)間戳轉(zhuǎn)換到同一個(gè)參考時(shí)區(qū)如UTC本身或某個(gè)標(biāo)準(zhǔn)時(shí)區(qū)。4. 核心分析利用Spark SQL與高級(jí)API挖掘氣象規(guī)律數(shù)據(jù)準(zhǔn)備就緒后就進(jìn)入了最核心的分析階段。Spark提供了多套APIRDD, DataFrame, SQL對(duì)于結(jié)構(gòu)化數(shù)據(jù)的分析Spark SQL和DataFrame API因其聲明式的風(fēng)格和強(qiáng)大的優(yōu)化能力是最高效的選擇。4.1 使用Spark SQL進(jìn)行靈活的時(shí)空查詢將DataFrame注冊(cè)為臨時(shí)視圖后就可以使用標(biāo)準(zhǔn)的SQL語法進(jìn)行查詢這對(duì)于熟悉SQL的數(shù)據(jù)分析師來說非常友好。# 將寬表DataFrame注冊(cè)為臨時(shí)視圖 wide_df.createOrReplaceTempView(weather_wide) # 示例1查詢某個(gè)特定站點(diǎn)例如北京站假設(shè)ID為‘CHM00054511’2023年的夏季6,7,8月最高氣溫 spark.sql( SELECT station_id, date, tmax FROM weather_wide WHERE station_id CHM00054511 AND YEAR(date) 2023 AND MONTH(date) IN (6, 7, 8) AND tmax IS NOT NULL ORDER BY tmax DESC LIMIT 10 ).show() # 示例2計(jì)算每個(gè)省份需要關(guān)聯(lián)站點(diǎn)元數(shù)據(jù)表stations的年平均氣溫 # 假設(shè)我們有一個(gè)站點(diǎn)元數(shù)據(jù)表包含station_id和province字段 stations_df spark.read.csv(hdfs://path/to/ghcnd-stations.txt, headerFalse, inferSchemaFalse) # ... 解析stations_df此處省略 ... stations_df.createOrReplaceTempView(stations) spark.sql( SELECT s.province, YEAR(w.date) as year, AVG(w.tavg) as avg_annual_temp FROM weather_wide w JOIN stations s ON w.station_id s.station_id WHERE w.tavg IS NOT NULL GROUP BY s.province, YEAR(w.date) ORDER BY year, province ).show()Spark SQL支持復(fù)雜的嵌套查詢、窗口函數(shù)、Common Table Expressions (CTEs)等功能非常強(qiáng)大。Catalyst優(yōu)化器會(huì)自動(dòng)對(duì)SQL語句進(jìn)行邏輯和物理優(yōu)化比如謂詞下推、列裁剪等即使面對(duì)海量數(shù)據(jù)也能保證較高的查詢效率。4.2. 窗口函數(shù)在時(shí)序分析中的高級(jí)應(yīng)用氣象數(shù)據(jù)是典型的時(shí)間序列數(shù)據(jù)。窗口函數(shù)是分析時(shí)間序列的利器可以方便地計(jì)算移動(dòng)平均、累計(jì)和、前后期對(duì)比等。from pyspark.sql.window import Window from pyspark.sql.functions import avg, sum as _sum, lag, row_number # 為每個(gè)站點(diǎn)定義時(shí)間窗口 window_spec Window.partitionBy(station_id).orderBy(date).rowsBetween(-6, 0) # 當(dāng)前及前6天共7天 # 計(jì)算7日移動(dòng)平均氣溫 df_with_ma wide_df.withColumn(tavg_7d_ma, avg(tavg).over(window_spec)) # 計(jì)算每個(gè)站點(diǎn)每年的高溫日數(shù)日最高溫超過35度 df_heatwave wide_df.withColumn(is_heatwave, when(col(tmax) 35, 1).otherwise(0)) annual_heatwave_days df_heatwave.groupBy(station_id, year(date).alias(year)) \ .agg(_sum(is_heatwave).alias(heatwave_days)) # 使用lag函數(shù)計(jì)算日際溫差 df_with_temp_change wide_df.withColumn(prev_day_tavg, lag(tavg, 1).over(Window.partitionBy(station_id).orderBy(date))) df_with_temp_change df_with_temp_change.withColumn(daily_temp_change, col(tavg) - col(prev_day_tavg))4.3. 利用MLlib進(jìn)行簡單的氣象預(yù)測(cè)與模式識(shí)別Spark MLlib是Spark的機(jī)器學(xué)習(xí)庫雖然不如Scikit-learn算法豐富但對(duì)于大規(guī)模數(shù)據(jù)集上的分布式訓(xùn)練有天然優(yōu)勢(shì)。我們可以嘗試一些基礎(chǔ)的預(yù)測(cè)任務(wù)。例如我們想基于過去幾天的天氣情況預(yù)測(cè)明天的最高氣溫。這是一個(gè)回歸問題。from pyspark.ml.feature import VectorAssembler, StandardScaler from pyspark.ml.regression import LinearRegression, RandomForestRegressor from pyspark.ml import Pipeline from pyspark.ml.evaluation import RegressionEvaluator # 1. 準(zhǔn)備特征使用過去3天的tmax, tmin, prcp作為特征 feature_cols [] for i in range(1, 4): # 滯后1天2天3天 for var in [tmax, tmin, prcp]: col_name f{var}_lag_{i} df_with_lags df_with_lags.withColumn(col_name, lag(var, i).over(Window.partitionBy(station_id).orderBy(date))) feature_cols.append(col_name) # 2. 定義標(biāo)簽明天的tmax df_with_lags df_with_lags.withColumn(label, lead(tmax, 1).over(Window.partitionBy(station_id).orderBy(date))) # 3. 過濾掉特征或標(biāo)簽為空的記錄 modeling_df df_with_lags.filter(col(label).isNotNull() ~(col(tmax_lag_1).isNull())) # 4. 特征向量化與標(biāo)準(zhǔn)化 assembler VectorAssembler(inputColsfeature_cols, outputColraw_features) scaler StandardScaler(inputColraw_features, outputColfeatures, withStdTrue, withMeanTrue) # 5. 劃分訓(xùn)練集和測(cè)試集 train_df, test_df modeling_df.randomSplit([0.8, 0.2], seed42) # 6. 構(gòu)建并訓(xùn)練模型以隨機(jī)森林為例 rf RandomForestRegressor(featuresColfeatures, labelCollabel, numTrees50, maxDepth10) pipeline Pipeline(stages[assembler, scaler, rf]) model pipeline.fit(train_df) # 7. 預(yù)測(cè)與評(píng)估 predictions model.transform(test_df) evaluator RegressionEvaluator(labelCollabel, predictionColprediction, metricNamermse) rmse evaluator.evaluate(predictions) print(fRoot Mean Squared Error (RMSE) on test data {rmse})經(jīng)驗(yàn)技巧在構(gòu)建時(shí)序特征時(shí)一定要注意數(shù)據(jù)泄露問題。絕對(duì)不能使用未來的信息來預(yù)測(cè)過去或現(xiàn)在。lag函數(shù)是安全的因?yàn)樗皇褂眠^去的數(shù)據(jù)而lead函數(shù)是用來生成標(biāo)簽的。在劃分訓(xùn)練集和測(cè)試集時(shí)更嚴(yán)謹(jǐn)?shù)淖龇ㄊ前磿r(shí)間劃分例如用2010-2019的數(shù)據(jù)訓(xùn)練用2020年的數(shù)據(jù)測(cè)試而不是隨機(jī)劃分因?yàn)樘鞖鈹?shù)據(jù)具有很強(qiáng)的時(shí)間自相關(guān)性。隨機(jī)劃分會(huì)破壞這種結(jié)構(gòu)導(dǎo)致評(píng)估結(jié)果過于樂觀。此外對(duì)于氣象預(yù)測(cè)更復(fù)雜的模型如LSTM神經(jīng)網(wǎng)絡(luò)可能效果更好但這通常需要將數(shù)據(jù)收集到驅(qū)動(dòng)節(jié)點(diǎn)并使用TensorFlow或PyTorch或者使用Spark的Deep Learning Pipelines等擴(kuò)展庫這超出了基礎(chǔ)MLlib的范圍。5. 性能調(diào)優(yōu)與生產(chǎn)化考量當(dāng)數(shù)據(jù)量和計(jì)算復(fù)雜度增長時(shí)默認(rèn)的Spark配置可能無法帶來最佳性能甚至?xí)霈F(xiàn)OOM或任務(wù)失敗。調(diào)優(yōu)是Spark作業(yè)從“能跑”到“跑得快且穩(wěn)”的關(guān)鍵一步。5.1. 理解Spark執(zhí)行計(jì)劃與數(shù)據(jù)傾斜診斷Spark UIWeb界面是你最好的朋友。任何性能調(diào)優(yōu)都應(yīng)從查看Spark UI開始。提交一個(gè)作業(yè)后訪問http://driver-host:4040對(duì)于應(yīng)用運(yùn)行期間或Spark History Server對(duì)于已完成的應(yīng)用。重點(diǎn)關(guān)注Stages和Executors標(biāo)簽頁Tasks的數(shù)量和持續(xù)時(shí)間一個(gè)Stage內(nèi)的所有Task執(zhí)行時(shí)間應(yīng)該大致相同。如果出現(xiàn)個(gè)別Task執(zhí)行時(shí)間極長長尾任務(wù)很可能遇到了數(shù)據(jù)傾斜。數(shù)據(jù)傾斜是指某個(gè)或某幾個(gè)Key對(duì)應(yīng)的數(shù)據(jù)量遠(yuǎn)大于其他Key導(dǎo)致處理這些Key的Task成為瓶頸。Shuffle讀寫量Shuffle數(shù)據(jù)混洗發(fā)生在groupBy、join、repartition等操作后是Spark中最昂貴的操作。過大的Shuffle數(shù)據(jù)量會(huì)拖慢整個(gè)作業(yè)。如何診斷數(shù)據(jù)傾斜可以在代碼中抽樣檢查Key的分布。# 檢查groupBy操作前的Key分布 key_counts df.groupBy(your_key_column).count().orderBy(col(count).desc()) key_counts.show(10) # 查看前10個(gè)最多的Key如果發(fā)現(xiàn)某個(gè)Key的數(shù)量級(jí)是其他的成百上千倍就確認(rèn)了傾斜。處理數(shù)據(jù)傾斜的常見策略過濾異常Key如果傾斜的Key是異常數(shù)據(jù)如測(cè)試數(shù)據(jù)、空值可以直接過濾掉。增加Shuffle分區(qū)數(shù)通過spark.sql.shuffle.partitions默認(rèn)200增加分區(qū)數(shù)讓大Key的數(shù)據(jù)分散到更多Task中。但這治標(biāo)不治本如果某個(gè)Key的數(shù)據(jù)量本身巨大增加分區(qū)可能無效。兩階段聚合對(duì)于聚合操作先給Key加上隨機(jī)前綴進(jìn)行局部聚合再去掉前綴進(jìn)行全局聚合。這能打散大Key。from pyspark.sql.functions import concat_ws, rand, col # 假設(shè)要對(duì)province進(jìn)行求和且province存在傾斜 df_with_salt df.withColumn(salted_key, concat_ws(_, col(province), (rand() * 10).cast(int).cast(string))) stage1 df_with_salt.groupBy(salted_key).agg(sum(value).alias(partial_sum)) # 去掉隨機(jī)后綴進(jìn)行最終聚合 stage1.withColumn(original_key, split(col(salted_key), _)[0]) \ .groupBy(original_key).agg(sum(partial_sum).alias(total_sum))使用廣播連接如果一個(gè)表非常小比如站點(diǎn)元數(shù)據(jù)表可以將其廣播到所有Executor避免Shuffle。使用broadcast提示。from pyspark.sql.functions import broadcast large_df.join(broadcast(small_df), station_id)5.2. 關(guān)鍵配置參數(shù)詳解與調(diào)優(yōu)實(shí)踐Spark有上百個(gè)配置參數(shù)但核心的只有十幾個(gè)。以下是一些在生產(chǎn)環(huán)境中經(jīng)常需要調(diào)整的spark.executor.memory和spark.driver.memory如前所述合理設(shè)置。通常Executor內(nèi)存的10%-20%會(huì)留給堆外內(nèi)存和系統(tǒng)開銷。spark.sql.shuffle.partitions控制Shuffle后的分區(qū)數(shù)。默認(rèn)200通常偏小對(duì)于大數(shù)據(jù)集可以設(shè)置為num_executors * num_cores_per_executor * 2~4。但分區(qū)數(shù)過多也會(huì)帶來調(diào)度開銷。spark.default.parallelism對(duì)于沒有父RDD/DataFrame的操作如從HDFS讀取其初始分區(qū)數(shù)。建議設(shè)置為集群總核心數(shù)的2-3倍。spark.sql.adaptive.enabled(AQE)自適應(yīng)查詢執(zhí)行是Spark 3.x的重大特性強(qiáng)烈建議開啟默認(rèn)在Spark 3.2是開啟的。它能動(dòng)態(tài)合并過小的分區(qū)、優(yōu)化傾斜的連接、在運(yùn)行時(shí)調(diào)整Join策略極大地簡化了手動(dòng)調(diào)優(yōu)的工作。spark.sql.autoBroadcastJoinThreshold控制自動(dòng)進(jìn)行廣播連接的表大小閾值單位字節(jié)。默認(rèn)10MB。如果你的小表有幾十MB且內(nèi)存充足可以適當(dāng)調(diào)大此值。spark.serializer使用org.apache.spark.serializer.KryoSerializer它比默認(rèn)的Java序列化更快、更緊湊。但需要注冊(cè)自定義類。一個(gè)典型的提交命令可能如下spark-submit \ --master spark://master:7077 \ --deploy-mode client \ --num-executors 10 \ --executor-cores 4 \ --executor-memory 8g \ --driver-memory 2g \ --conf spark.sql.shuffle.partitions400 \ --conf spark.sql.adaptive.enabledtrue \ --conf spark.serializerorg.apache.spark.serializer.KryoSerializer \ your_weather_analysis_job.py5.3. 數(shù)據(jù)持久化策略與內(nèi)存管理在復(fù)雜的作業(yè)中一個(gè)DataFrame可能會(huì)被多次使用例如在特征工程的不同階段。每次行動(dòng)操作如count(),show(),write都會(huì)觸發(fā)從頭計(jì)算。為了避免重復(fù)計(jì)算需要將中間結(jié)果持久化緩存。# 緩存DataFrame到內(nèi)存和磁盤 processed_df.persist(storageLevelStorageLevel.MEMORY_AND_DISK_SER) # 或者使用快捷方法 processed_df.cache() # 等同于 MEMORY_ONLY_SER # 觸發(fā)一個(gè)行動(dòng)操作真正開始緩存 processed_df.count() # ... 后續(xù)使用processed_df的多個(gè)操作 ... # 作業(yè)結(jié)束后釋放緩存 processed_df.unpersist()選擇正確的存儲(chǔ)級(jí)別很重要MEMORY_ONLY只存內(nèi)存最快但如果內(nèi)存不夠分區(qū)會(huì)被重新計(jì)算。MEMORY_AND_DISK優(yōu)先存內(nèi)存內(nèi)存不夠時(shí)溢寫到磁盤。這是最常用的平衡選擇。MEMORY_ONLY_SER/MEMORY_AND_DISK_SER序列化后存儲(chǔ)更省內(nèi)存但讀寫時(shí)需要序列化/反序列化開銷。避坑指南不要無腦緩存所有中間DataFrame。緩存會(huì)占用寶貴的集群內(nèi)存。只緩存那些確實(shí)會(huì)被多次使用且計(jì)算成本高昂的DataFrame。一個(gè)常見的反模式是在一個(gè)循環(huán)中反復(fù)讀取和緩存同一個(gè)數(shù)據(jù)源。另外記得在不再需要時(shí)調(diào)用unpersist()尤其是在長時(shí)間運(yùn)行的Spark Streaming應(yīng)用中否則會(huì)導(dǎo)致內(nèi)存泄漏。對(duì)于迭代式機(jī)器學(xué)習(xí)算法如ALS推薦Spark MLlib會(huì)自動(dòng)處理RDD的持久化通常不需要手動(dòng)干預(yù)。最后警惕廣播變量的大小。雖然廣播變量很方便但如果廣播一個(gè)巨大的數(shù)據(jù)集比如幾百M(fèi)B甚至上GB會(huì)消耗大量Driver和Executor的網(wǎng)絡(luò)帶寬和內(nèi)存可能導(dǎo)致Driver OOM。通常建議廣播變量的大小不要超過幾百M(fèi)B。