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