簡介面向大數(shù)據(jù)期末課設(shè)的Spark氣象數(shù)據(jù)處理與分析完整報告適合需要完成Spark實戰(zhàn)項目、掌握氣象數(shù)據(jù)分析全流程的大數(shù)據(jù)專業(yè)學(xué)生參考。項目基于中央氣象臺2412個城市、57888條真實天氣數(shù)據(jù)完整呈現(xiàn)從JSON異步數(shù)據(jù)爬取、CSV存儲、Spark DataFrame清洗聚合到matplotlib可視化的實現(xiàn)過程并給出省份、城市、天氣三級接口的解析思路與爬蟲源代碼。報告按課程設(shè)計標(biāo)準(zhǔn)撰寫章節(jié)涵蓋項目背景、實驗環(huán)境、數(shù)據(jù)來源、數(shù)據(jù)獲取、數(shù)據(jù)分析與可視化其中重點演示了分組聚合、缺失異常值處理、窗口函數(shù)等Spark常用操作可作為課程報告或答辯展示的實用模板。內(nèi)含1個docx文檔744KB已有13405人學(xué)習(xí)適合大數(shù)據(jù)入門者對照演練。1. 基于Spark的氣象數(shù)據(jù)處理與分析一份能直接復(fù)用的大數(shù)據(jù)期末課設(shè)每年期末總有同學(xué)卡在同一個問題上Spark課設(shè)到底做什么才能既體現(xiàn)技術(shù)含量又能在兩周內(nèi)跑完這份基于Spark的氣象數(shù)據(jù)處理與分析課設(shè)項目走的就是一條非常務(wù)實的路線——爬取全國2412個城市、57888條整點天氣數(shù)據(jù)用Spark SQL完成24小時累計雨量和日均氣溫的聚合計算再用matplotlib把結(jié)果畫出來。整套流程覆蓋了數(shù)據(jù)獲取、清洗、分析、可視化四個環(huán)節(jié)技術(shù)棧就是Spark DataFrame API加Python沒有繞圈子也沒有堆砌高深算法。適合大數(shù)據(jù)專業(yè)期末課設(shè)、Spark入門實戰(zhàn)以及想了解“用Spark做氣象數(shù)據(jù)處理”完整鏈路的人。下文我按實際落地順序把環(huán)境、爬蟲、預(yù)處理、分析和踩坑點逐一拆開講。2. 環(huán)境與數(shù)據(jù)獲取從三級URL到57888條天氣數(shù)據(jù)的爬取鏈路2.1 實驗環(huán)境選型與組件安裝這個課設(shè)的實驗環(huán)境是一個典型的單機Spark開發(fā)環(huán)境不需要搭建集群。我復(fù)現(xiàn)時使用的組件版本如下表所示組件版本說明LinuxUbuntu 16.04課設(shè)報告中指定的操作系統(tǒng)Python3.9用于爬蟲、Spark任務(wù)和可視化Spark2.4.0計算引擎跑在local模式下Jupyter Notebook最新版交互式調(diào)試DataFrame和繪圖有一點值得注意Spark 2.4.0與Python 3.9的組合在實際使用中會有兼容性警告網(wǎng)上很多資料說Spark 2.4.0不支持Python 3.8以上的版本實際跑起來Spark SQL基礎(chǔ)功能問題不大但如果你用到pandas UDF或新的API建議直接換Spark 3.x。課設(shè)場景下沿用2.4.0沒有問題只要別在它上面做太新奇的特性就行。裝完Spark和Jupyter之后還需要給Python補上可視化的依賴sudo apt-get install python3-matplotlib sudo apt-get install python3-tkpython3-matplotlib是繪圖庫本體python3-tk是matplotlib在本地窗口顯示圖表時需要的Tkinter后端。如果你用的是無顯示器的服務(wù)器還應(yīng)在代碼開頭加上matplotlib.use(Agg)否則繪圖時找不到顯示設(shè)備會直接報錯。Ubuntu 16.04的apt源里自帶這兩個包優(yōu)先用它裝比pip裝更省事。2.2 三級URL接口設(shè)計先拿省份再拿城市最后拿天氣氣象數(shù)據(jù)的來源是某公開氣象服務(wù)站點。通過瀏覽器開發(fā)者工具觀察網(wǎng)絡(luò)請求能發(fā)現(xiàn)數(shù)據(jù)不是渲染在HTML頁面里的而是通過異步請求直接返回JSON字符串。核心的請求路徑可以歸納為下面三條層級請求路徑返回內(nèi)容省份列表/f/rest/province省份名稱與省份三位編碼城市列表/f/rest/province/{省份三位編碼}該省所有城市的名稱與城市編號整點天氣/f/rest/passed/{城市編號}該城市最近24小時整點天氣數(shù)組三級接口是典型的嵌套依賴關(guān)系省份編碼必須從第一級接口的返回里取城市編號必須從第二級接口的返回里取第三級接口才能拿到真正的氣象數(shù)據(jù)。比如某個省在省份接口里的編碼是“ABC”請求城市列表時要帶上這個編碼返回的城市數(shù)據(jù)里又有類似“58847”的城市編號把它拼到第三級URL末尾才能拿到該城市24小時整點數(shù)據(jù)。爬蟲就是要順著這條鏈子把全國城市依次走一遍。2.3 爬蟲代碼實現(xiàn)與CSV分段落盤爬蟲部分原課設(shè)報告用的是Python 2時期的urllib3.Request寫法但urllib3庫里并沒有Request類直接跑會報AttributeError。我改成Python 3標(biāo)準(zhǔn)庫的urllib.request邏輯保持一致import urllib.request import json import csv import codecs import os import time class WeatherCrawler: def __init__(self, base_url): self.base_url base_url def fetch_json(self, path): url self.base_url path req urllib.request.Request(url, headers{User-Agent: Mozilla/5.0}) with urllib.request.urlopen(req, timeout10) as resp: return json.loads(resp.read().decode(utf-8)) def write_csv(self, file_name, rows, write_headerFalse): if not rows: return with open(file_name, ab) as f: f.write(codecs.BOM_UTF8) writer csv.DictWriter(f, fieldnamesrows[0].keys()) if write_header: writer.writeheader() writer.writerows(rows)fetch_json里設(shè)置了User-Agent請求頭避免站點拒絕默認(rèn)的Python urllib請求。timeout10是超時控制防止某個城市接口卡住導(dǎo)致整個爬蟲掛起。write_csv里寫入codecs.BOM_UTF8是為了讓Excel直接打開CSV時中文不亂碼write_header參數(shù)控制是否寫表頭因為分段落盤時只有第一個文件塊需要寫表頭。下面是三級數(shù)據(jù)獲取的完整調(diào)度邏輯def get_provinces(self): if os.path.exists(input/province.csv): return self.read_csv(input/province.csv) provinces self.fetch_json(/f/rest/province) self.write_csv(input/province.csv, provinces, write_headerTrue) return provinces def get_cities(self): if os.path.exists(input/city.csv): return self.read_csv(input/city.csv) cities [] for province in self.get_provinces(): code province[url].split(/)[-1].split(.)[0] cities.extend(self.fetch_json(/f/rest/province/ code)) time.sleep(1) self.write_csv(input/city.csv, cities, write_headerTrue) return cities def get_passed_weather(self): cities self.get_cities() passed_weather [] count 0 for city in cities: data self.fetch_json(/f/rest/passed/ city[code]) if data: count 1 for item in data: item[city_code] city[code] item[province] city[province] item[city_name] city[city] item[city_index] str(count) passed_weather.extend(data) time.sleep(1) if count % 50 0 and count 0: self.write_csv(input/passed_weather_ALL.csv, passed_weather, write_header(count 50)) passed_weather [] if passed_weather: self.write_csv(input/passed_weather_ALL.csv, passed_weather, write_header(count 50))這段代碼有兩個關(guān)鍵設(shè)計。第一個是冪等province.csv和city.csv已存在時直接讀取全量天氣數(shù)據(jù)已經(jīng)落盤后重新運行也不會重復(fù)爬取想刷新數(shù)據(jù)就手動刪掉這三個文件。第二個是每爬50個城市就落盤一次——全國2412個城市每條數(shù)據(jù)有多個整點記錄如果全部堆在內(nèi)存里等到最后再一次性寫文件爬半小時內(nèi)存早就爆了。分段寫入之后即使中間某個城市請求超時導(dǎo)致中斷已爬的部分也不會丟。3. Spark SQL數(shù)據(jù)預(yù)處理Schema推斷、類型轉(zhuǎn)換與異常過濾3.1 SparkSession初始化與CSV讀取數(shù)據(jù)爬下來之后下一步就是用Spark讀進來。這里用的是Spark SQL中的DataFrame API而不是RDD——DataFrame有類型信息SQL風(fēng)格表達清晰而且Spark默認(rèn)的Catalyst優(yōu)化器會自動優(yōu)化執(zhí)行計劃對課設(shè)這種量級的數(shù)據(jù)完全夠用。from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(WeatherAnalysis) \ .master(local[4]) \ .getOrCreate() df spark.read.csv( input/passed_weather_ALL.csv, headerTrue, inferSchemaTrue )master(local[4])表示本地模式用4個線程并行執(zhí)行對六萬條數(shù)據(jù)來說綽綽有余。headerTrue把CSV第一行作為列名inferSchemaTrue讓Spark掃描數(shù)據(jù)后自動推斷每一列的類型。這樣temperature和rain1h會被識別成double后續(xù)做數(shù)值計算不用手動轉(zhuǎn)換。但inferSchema有一個坑如果CSV某列的前面一部分行是缺失值或非數(shù)字字符串Spark可能把整列推斷成string后面用比較運算符直接報錯。穩(wěn)妥的做法是在讀取后打印Schema檢查一遍df.printSchema() df.show(5, truncateFalse)printSchema()輸出的每一行格式是“列名 類型”一眼就能看出有沒有推斷錯誤。show(5, truncateFalse)則是查看前5條原始數(shù)據(jù)truncateFalse確保長字段不會被截斷方便直接觀察time字段的格式。3.2 字段選擇與數(shù)值類型轉(zhuǎn)換原課設(shè)報告里有一個很容易被忽略的細節(jié)CSV經(jīng)過Python爬蟲寫出后即使Spark用inferSchema讀入某些列的類型仍是字符串。更穩(wěn)妥的做法是在分析前顯式做一次類型轉(zhuǎn)換把注意力集中在真正需要的列上from pyspark.sql.functions import col df_rain df.select( province, city_name, city_code, rain1h ).filter( col(rain1h).cast(double) 1000 )select只取四個字段把數(shù)據(jù)寬度壓縮到最小col(rain1h).cast(double)強制把降雨量轉(zhuǎn)成數(shù)值型這樣即使某一行是空字符串也不會讓整個作業(yè)報錯而是轉(zhuǎn)為null。過濾條件 1000是排除異常值——正常城市24小時累計降雨量不可能超過1000毫米一旦出現(xiàn)這種記錄基本是站點數(shù)據(jù)采集故障或爬取時拿到的臟數(shù)據(jù)。3.3 時間字段解析與溫度數(shù)據(jù)準(zhǔn)備溫度分析需要把time字段拆成“日期”和“小時”兩部分這就要用到pyspark.sql.functions里的兩個函數(shù)from pyspark.sql.functions import date_format, hour df_temp df.select( province, city_name, city_code, temperature, time ).withColumn( date, date_format(col(time), yyyy-MM-dd) ).withColumn( hour, hour(col(time)) )date_format(col(time), yyyy-MM-dd)從時間戳里截取日期部分hour(col(time))提取小時數(shù)值。這里的前提是CSV里time字段能被Spark識別為時間類型。如果Spark讀進來之后time是字符串這兩個函數(shù)會返回null。遇到這種情況先看原始數(shù)據(jù)的time字段格式再決定是加一個to_timestamp(col(time), yyyy-MM-dd HH:mm:ss)轉(zhuǎn)換還是直接用字符串截取函數(shù)處理。整個預(yù)處理階段的目標(biāo)就是得到兩張干凈的中間表一張是帶rain1h數(shù)值的df_rain一張是帶date和hour字段的df_temp。兩張都是懶加載的DataFrame真正觸發(fā)計算的是后續(xù)的groupBy聚合操作。4. 兩個核心分析任務(wù)24小時累計雨量與四時次日均氣溫4.1 累計降雨量分析groupBy、agg與全局排序降雨量分析的思路很直接按省份和城市分組對rain1h字段求和得到每個城市過去24小時的累計雨量然后降序排列取Top-N。Spark代碼是這樣的df_rain_sum df_rain.groupBy( province, city_name, city_code ).agg( {rain1h: sum} ).withColumnRenamed( sum(rain1h), rain24h ).sort( col(rain24h).desc() )agg({rain1h: sum})是傳入字典的寫法鍵是列名值是聚合函數(shù)結(jié)果列名自動生成為sum(rain1h)所以緊接著用withColumnRenamed重命名成rain24h后面引用起來更直觀。.sort(col(rain24h).desc())是分布式環(huán)境下的全局排序Spark會做一次shuffle把所有分區(qū)數(shù)據(jù)按雨量倒排對2400多個城市的數(shù)據(jù)量來說代價很小。排序之后head(10)可以直接取到雨量最大的十個城市這個動作會觸發(fā)Spark真正執(zhí)行前面的所有轉(zhuǎn)換步驟top_rain df_rain_sum.head(10) for row in top_rain: print(row[city_name], row[rain24h])注意head(10)返回的是一個Row對象列表每個Row就像是Python里的字典用row[列名]取值。這一步執(zhí)行時Spark會把懶加載的整個血緣關(guān)系鏈從CSV讀取一路算到排序所以如果前面某一步寫錯了報錯是在這里炸出來而不是在groupBy那里。4.2 日均氣溫分析四時次篩選與聚合計算日均氣溫這一段是項目里最有氣象業(yè)務(wù)味道的地方。根據(jù)《地面氣象服務(wù)觀測規(guī)范》日平均氣溫取的是02時、08時、14時、20時四個時次氣溫的平均值而不是24小時整點氣溫的算術(shù)平均。這是氣象行業(yè)的規(guī)范直接決定了代碼里的過濾條件from pyspark.sql.functions import count, sum as spark_sum, format_number df_4point df_temp.filter( col(hour).isin([2, 8, 14, 20]) ) df_avg df_4point.groupBy( province, city_name, city_code, date ).agg( count(temperature).alias(cnt), spark_sum(temperature).alias(temp_sum) ).filter( col(cnt) 4 ).withColumn( avg_temperature, format_number(col(temp_sum) / 4.0, 1) ).select( province, city_name, city_code, date, avg_temperature )整個流水線的關(guān)鍵是先計數(shù)再求均值count(temperature)看看每個城市在目標(biāo)日期里是否四個時次的數(shù)據(jù)都齊了只有cnt 4的記錄才參與平均計算。如果某個城市某天只有3個時次的數(shù)據(jù)直接算平均氣溫會和真實值有偏差過濾掉才是嚴(yán)謹(jǐn)?shù)淖龇?。format_number(..., 1)把結(jié)果格式化成帶一位小數(shù)的字符串方便直接用于可視化展示。這個查重邏輯在課設(shè)報告里已經(jīng)說得很細但我復(fù)現(xiàn)時發(fā)現(xiàn)一個數(shù)據(jù)層面的問題爬蟲是在白天某個時刻運行的當(dāng)天20時還沒到那么當(dāng)天20時的記錄根本不存在。原報告也提示了“建議在當(dāng)天20時30分之后再爬取數(shù)據(jù)”這說明分析結(jié)果的質(zhì)量高度依賴爬取時機。4.3 cache與coalesce性能優(yōu)化和輸出控制兩個分析任務(wù)在最后都調(diào)了cache()和coalesce(1)這一步值得展開講df_rain_sum.cache().coalesce(1).write.csv(output/rain_sum, headerTrue) df_avg.coalesce(1).write.json(output/avg_temp, headerFalse)cache()把前面經(jīng)過groupBy和sort的中間結(jié)果物化到內(nèi)存里這樣后面如果既要做head(10)又要write.csvSpark不用從頭再算兩遍。coalesce(1)是把分區(qū)數(shù)量壓縮到1個這樣CSV輸出只有一個文件而不是默認(rèn)的幾十個part-00000碎片后續(xù)直接讀這個單文件就行。但coalesce(1)不是免費的。它本質(zhì)上是一次結(jié)果重分區(qū)把所有分區(qū)的數(shù)據(jù)拉到同一個分區(qū)再寫出數(shù)據(jù)量大時會產(chǎn)生單點瓶頸。雨量Top-N分析加coalesce(1)沒問題但如果以后換成每天幾十GB的氣象數(shù)據(jù)正確做法是保留分區(qū)數(shù)并寫成parquet格式不要強行合并成單文件。這個邊界要拎清楚。5. 常見問題與排查爬蟲、Spark和可視化里的五個坑5.1 Python 3下運行報錯sys.setdefaultencoding不存在現(xiàn)象把課設(shè)原爬蟲代碼直接丟到Python 3環(huán)境運行啟動就報AttributeError: module sys has no attribute setdefaultencoding。原因sys.setdefaultencoding(utf8)是Python 2時代處理默認(rèn)編碼的命令Python 3的默認(rèn)編碼就是UTF-8這個函數(shù)已經(jīng)被移除。原課設(shè)寫的是reload(sys)和sys.setdefaultencoding(utf8)這是典型的Python 2代碼。解決直接刪掉這兩行。如果擔(dān)心中文編碼有問題用Python 3在打開文件時指定encodingutf-8就夠了不需要再動全局默認(rèn)編碼。這個坑幾乎是每屆課設(shè)必現(xiàn)看到這個報錯就知道代碼是從舊版本遷移過來的。5.2 爬蟲結(jié)果CSV用Excel打開中文亂碼現(xiàn)象passed_weather_ALL.csv在終端和代碼里讀著都正常用Excel雙擊打開時中文全部變成亂碼省份名和城市名完全不可讀。原因CSV文件沒有BOM頭。Python默認(rèn)用UTF-8寫文本文件不摻BOM而Excel的CSV解析器在識別非UTF-8編碼時傾向用ANSI解碼中文結(jié)果就是亂成一團。原報告里在write_csv中寫入codecs.BOM_UTF8就是為了解決這個問題。解決寫入CSV前先寫一次codecs.BOM_UTF8這是給Excel看的編碼標(biāo)記。注意每次打開文件追加寫入時都要判斷文件是否已存在如果已存在就不要重復(fù)加BOM否則表頭里會多出隱藏的非法字符。5.3 Spark讀CSV后數(shù)值列變成字符串過濾直接報錯現(xiàn)象df.filter(col(rain1h) 1000)執(zhí)行時拋出分析異常提示無法用字符串和數(shù)字比較打印Schema發(fā)現(xiàn)rain1h列的類型是string而非double。原因inferSchemaTrue不是萬能的。如果CSV的前若干行里rain1h字段是空值或占位符比如--Spark在推斷類型時會認(rèn)為整列是字符串。尤其是爬蟲寫CSV時把數(shù)字轉(zhuǎn)成了字符串空字段就是空字符串更容易觸發(fā)誤判。解決最直接的辦法是用col(rain1h).cast(double)做顯式轉(zhuǎn)換再配合filter過濾。另一種做法是給read.csv顯式傳schema參數(shù)例如StructType里指定rain1h為DoubleType()徹底跳過推斷機制。從工程習(xí)慣來說顯式指定Schema比inferSchema更可靠但課設(shè)階段寫清楚cast就夠了。5.4 日均氣溫結(jié)果城市數(shù)大幅縮水只有幾百個城市現(xiàn)象計算完日均氣溫后輸出結(jié)果里的城市數(shù)遠少于數(shù)據(jù)源里的2412個城市少了一半還多。檢查原始數(shù)據(jù)發(fā)現(xiàn)每個城市確實有整點氣溫數(shù)據(jù)。原因filter(col(cnt) 4)把只擁有部分時次記錄的城市全部剔除了。爬蟲運行時間是白天當(dāng)天20時的數(shù)據(jù)不存在有些城市02時或08時的記錄缺失湊不齊四時次的都被過濾掉了。解決四種選擇。一是按原報告建議在當(dāng)天20時30分之后再爬取保證當(dāng)天四個時次都齊二是把cnt 4放寬到cnt 3并在報告里說明“基于三時次近似計算日均氣溫”適合只想跑通流程的情況三是查看缺失時次的分布規(guī)律找出是哪幾個時次最容易缺四是修改爬蟲邏輯補充歷史數(shù)據(jù)接口而不是只爬最近24小時。課設(shè)報告采用的是第一種輸出的質(zhì)量是最高的。5.5 結(jié)果目錄里出現(xiàn)一堆part文件沒有單個CSV文件現(xiàn)象執(zhí)行df_rain_sum.write.csv(output/rain_sum)后打開目錄看到的是part-00000-xxx.csv、part-00001-xxx.csv十幾個文件而不是想象中的rain_sum.csv。原因Spark分布式寫入時每個分區(qū)各自寫一個文件分區(qū)數(shù)就是輸出文件數(shù)。默認(rèn)情況下shuffle之后分區(qū)數(shù)較多自然會產(chǎn)生多個part文件。這不算錯誤但對后續(xù)人工查看數(shù)據(jù)很不方便。解決寫之前加coalesce(1)強制把分區(qū)數(shù)減到1輸出就只有一個part文件了。更省事的辦法是不加coalesce(1)讀取時直接spark.read.csv(output/rain_sum)讀整個目錄Spark會自動合并多個文件的內(nèi)容。兩種方式我都用過課設(shè)評分時如果要求提交“干凈的CSV文件”就選coalesce(1)。6. 結(jié)果驗證與可視化輸出用matplotlib把分析結(jié)果變成圖表數(shù)據(jù)分析算完不算結(jié)束還要驗證結(jié)果正確然后再畫圖。驗證有一個很土但很有效的辦法——抽樣對賬。我從df_rain_sum.head(10)里隨機挑了三個城市找到它們在passed_weather_ALL.csv里的所有rain1h記錄手工在Excel里求和和Spark算出的rain24h對比。三次都對得上說明聚合邏輯沒問題。這一步強烈建議保留它能把“代碼跑通”和“結(jié)果正確”區(qū)分開答辯時也更有底氣?？梢暬糠执a比較簡短用matplotlib畫橫向條形圖展示雨量Top-10城市import matplotlib.pyplot as plt top df_rain_sum.head(10) names [row[city_name] for row in top] values [float(row[rain24h]) for row in top] plt.rcParams[font.sans-serif] [SimHei, WenQuanYi Zen Hei] plt.rcParams[axes.unicode_minus] False plt.figure(figsize(10, 6)) plt.barh(names[::-1], values[::-1]) plt.xlabel(24小時累計降雨量mm) plt.title(過去24小時累計降雨量 Top10 城市) plt.tight_layout() plt.savefig(output/rain_top10.png, dpi150)plt.rcParams兩行是解決中文字符顯示成方塊的設(shè)置SimHei是Windows下的黑體Linux下沒有時就回退到WenQuanYi Zen Hei。names[::-1]和values[::-1]是把列表倒序讓雨量最大的城市顯示在條形圖最上方符合閱讀習(xí)慣。savefig的dpi150保證圖片在答辯投屏?xí)r不模糊。氣溫分析的結(jié)果可以用類似思路畫一張散點圖或直方圖查看全國城市日均氣溫的分布形態(tài)。到這一步從爬蟲、清洗、聚合到出圖整條鏈路就閉環(huán)了。想起那次復(fù)現(xiàn)后半段我發(fā)現(xiàn)某個城市計算出日均氣溫為97度一查是20時的氣溫字段爬到了異常值溫度列混入了噪點數(shù)據(jù)。從那以后我每次做數(shù)據(jù)類課設(shè)都會強制走一遍“抽樣對賬”再進入可視化這個習(xí)慣幫我擋掉了至少三次答辯翻車。希望這份基于Spark的氣象數(shù)據(jù)處理與分析課設(shè)梳理能幫你也少踩幾個同樣的坑。本文還有配套的精品資源點擊獲取