簡(jiǎn)介基于Spark的地鐵大數(shù)據(jù)客流分析系統(tǒng)畢業(yè)設(shè)計(jì)源碼包面向計(jì)算機(jī)專業(yè)學(xué)生及大數(shù)據(jù)初學(xué)者可支撐課程設(shè)計(jì)或畢設(shè)中的客流統(tǒng)計(jì)、預(yù)測(cè)與優(yōu)化分析。壓縮包內(nèi)主目錄按文檔、源碼、配置、數(shù)據(jù)等模塊組織共194個(gè)文件、約42.6MB以Java/Scala源碼、PNG截圖、XML配置、SQL腳本及Markdown文檔為主涵蓋需求分析、系統(tǒng)設(shè)計(jì)、代碼實(shí)現(xiàn)、測(cè)試數(shù)據(jù)與結(jié)果展示等完整環(huán)節(jié)。資源內(nèi)容涉及Spark SQL與Spark Streaming、MLlib、HBase及MySQL數(shù)據(jù)庫等知識(shí)點(diǎn)包含Spark作業(yè)腳本、數(shù)據(jù)庫建表腳本、接口定義并附有數(shù)據(jù)采集配置、Nginx日志配置和地鐵客流CSV樣本數(shù)據(jù)便于理解從數(shù)據(jù)采集、清洗、分析到可視化展示的全鏈路開發(fā)思路。已有259人學(xué)習(xí)瀏覽適合作為畢業(yè)設(shè)計(jì)模板或大數(shù)據(jù)綜合實(shí)訓(xùn)參考既可用于快速復(fù)現(xiàn)也可在此基礎(chǔ)上擴(kuò)展創(chuàng)新功能。1. 地鐵客流分析畢設(shè)怎么做從一堆日志到一張預(yù)測(cè)曲線做基于Spark的地鐵大數(shù)據(jù)客流分析系統(tǒng)這個(gè)畢設(shè)時(shí)最頭疼的不是寫代碼而是不知道該從哪里下手。某地鐵運(yùn)營(yíng)方每天都產(chǎn)生幾十GB的進(jìn)出站日志和刷卡流水他們想知道的不只是“今天多少人坐地鐵”而是“早高峰哪個(gè)站壓力最大、下個(gè)小時(shí)某站會(huì)進(jìn)多少人、要不要加開列車”。這些需求單靠MySQL和Excel根本算不動(dòng)而完整跑通一套從數(shù)據(jù)采集、清洗、存儲(chǔ)到預(yù)測(cè)的系統(tǒng)恰好就是大數(shù)據(jù)畢設(shè)的最佳閉環(huán)。這套方案適合正在做大數(shù)據(jù)類畢業(yè)設(shè)計(jì)的學(xué)生也適合想快速搭建基于Spark的離線與實(shí)時(shí)分析demo的開發(fā)者。它解決了畢設(shè)里最常見的問題需求太散、技術(shù)棧太雜、驗(yàn)收沒標(biāo)準(zhǔn)。下面按一套可落地的方案逐層拆開講。2. 需求分析與技術(shù)選型為什么核心是Spark而不是單純MySQL2.1 功能需求拆解客流統(tǒng)計(jì)、趨勢(shì)分析、峰值預(yù)測(cè)客流分析系統(tǒng)首先要回答三類問題過去發(fā)生了什么、現(xiàn)在正在發(fā)生什么、接下來可能發(fā)生什么。對(duì)應(yīng)到畢設(shè)里就是三個(gè)核心功能模塊離線統(tǒng)計(jì)、實(shí)時(shí)監(jiān)測(cè)、短期預(yù)測(cè)。離線統(tǒng)計(jì)解決“過去發(fā)生了什么”。按站點(diǎn)、線路、時(shí)間段統(tǒng)計(jì)進(jìn)站量、出站量和斷面客流量比如“某線某站早7點(diǎn)到9點(diǎn)的進(jìn)站人數(shù)”“全網(wǎng)昨日總客流”。這部分用Spark SQL跑批處理結(jié)果寫入數(shù)據(jù)庫供報(bào)表查詢。實(shí)時(shí)監(jiān)測(cè)解決“當(dāng)前發(fā)生了什么”統(tǒng)計(jì)最近5分鐘或10分鐘內(nèi)各站進(jìn)出站人數(shù)超過閾值就預(yù)警對(duì)應(yīng)Spark Streaming或Structured Streaming。短期預(yù)測(cè)解決“接下來會(huì)發(fā)生什么”基于歷史客流序列用MLlib的回歸算法預(yù)測(cè)未來一小時(shí)各站客流為排班和調(diào)度提供參考。這是典型的畢設(shè)級(jí)需求三個(gè)模塊各有各的數(shù)據(jù)流但都能用項(xiàng)目文件里的測(cè)試數(shù)據(jù)驗(yàn)證。功能不貪多把這三個(gè)場(chǎng)景做到能跑、能看、能講答辯就站穩(wěn)了。2.2 技術(shù)選型Spark三件套配合HBase做列式存儲(chǔ)選擇Spark作為核心最直接的原因是數(shù)據(jù)規(guī)模。地鐵閘機(jī)每臺(tái)每秒產(chǎn)生多條記錄全網(wǎng)一天的數(shù)據(jù)量足以讓單機(jī)MySQL的查詢響應(yīng)變得難以忍受。Spark基于內(nèi)存的分布式計(jì)算模型適合這種需要反復(fù)迭代和跨時(shí)間段聚合的統(tǒng)計(jì)場(chǎng)景。具體到組件選型Spark SQL負(fù)責(zé)離線結(jié)構(gòu)化查詢替代Hive或MapReduce開發(fā)效率高且支持DataFrame APISpark Streaming負(fù)責(zé)實(shí)時(shí)批次處理從Kafka消費(fèi)日志并按時(shí)間窗口聚合MLlib負(fù)責(zé)客流預(yù)測(cè)模型訓(xùn)練線性回歸和決策樹在中等數(shù)據(jù)量下足夠用。存儲(chǔ)層采用HBase加MySQL的雙存儲(chǔ)策略HBase存時(shí)序客流明細(xì)和聚合結(jié)果MySQL存站點(diǎn)、線路等維度信息。之所以不用純MySQL存明細(xì)是因?yàn)榭土髅骷?xì)按時(shí)間維度持續(xù)追加HBase的列式存儲(chǔ)和行鍵設(shè)計(jì)對(duì)這種時(shí)序?qū)懭敫押脪呙枘硞€(gè)時(shí)間段的數(shù)據(jù)也更快。選型對(duì)比可以從這個(gè)角度理解MapReduce能算但慢每輪作業(yè)都要落盤Spark把中間結(jié)果放內(nèi)存迭代計(jì)算快一個(gè)量級(jí)。HBase與MySQL對(duì)比也有清晰邊界。對(duì)比項(xiàng)MySQLHBase數(shù)據(jù)模型行式存儲(chǔ)強(qiáng)事務(wù)列式存儲(chǔ)稀疏表適合場(chǎng)景維度數(shù)據(jù)、低頻更新時(shí)序追加、范圍掃描擴(kuò)展方式主從/分庫分表Region自動(dòng)分裂在本系統(tǒng)的職責(zé)站點(diǎn)線路等基礎(chǔ)信息客流明細(xì)與聚合結(jié)果這套選型不是炫技它和畢設(shè)答辯時(shí)“為什么不用XXX”這類問題是直接對(duì)應(yīng)的回答時(shí)把上面這張表的邏輯講出來技術(shù)選型環(huán)節(jié)就過關(guān)了。2.3 數(shù)據(jù)來源與測(cè)試數(shù)據(jù)一份CSV如何撐起整條數(shù)據(jù)管道真實(shí)的地鐵刷卡數(shù)據(jù)屬于運(yùn)營(yíng)方內(nèi)部數(shù)據(jù)畢設(shè)拿不到所以項(xiàng)目里用一份地鐵客流CSV文件作為模擬數(shù)據(jù)源。這個(gè)CSV包含某城市地鐵網(wǎng)絡(luò)一段時(shí)間內(nèi)每條進(jìn)出站記錄字段設(shè)計(jì)為記錄時(shí)間戳、線路編號(hào)、站點(diǎn)ID、進(jìn)出站標(biāo)志、閘機(jī)編號(hào)、客流人數(shù)。每一行代表一次閘機(jī)通行測(cè)試數(shù)據(jù)量按幾萬到幾十萬條設(shè)計(jì)既能驗(yàn)證Spark分布式計(jì)算能力又不至于像真實(shí)幾十GB數(shù)據(jù)那樣讓本地開發(fā)環(huán)境跑不動(dòng)。這份CSV承擔(dān)兩個(gè)角色。第一是離線分析的輸入Spark SQL直接讀取CSV做清洗和統(tǒng)計(jì)。第二是實(shí)時(shí)管道的模擬數(shù)據(jù)源通過腳本按時(shí)間回放CSV內(nèi)容模擬Kafka中持續(xù)到達(dá)的客流消息讓實(shí)時(shí)模塊在沒有真實(shí)Nginx日志的情況也能完整跑通。3. 總體架構(gòu)設(shè)計(jì)從Nginx日志到HBase存儲(chǔ)的四層模塊劃分3.1 分層架構(gòu)采集層、計(jì)算層、存儲(chǔ)層、應(yīng)用層怎么協(xié)作整套系統(tǒng)按數(shù)據(jù)流向分成四層每一層職責(zé)單一方便獨(dú)立部署和調(diào)試。采集層用Nginx模擬地鐵閘機(jī)請(qǐng)求入口生成訪問日志Logstash監(jiān)聽日志文件按grok規(guī)則解析出時(shí)間戳、站點(diǎn)、進(jìn)出站等字段并發(fā)往Kafka作消息緩沖。計(jì)算層是核心一個(gè)Spark Streaming作業(yè)從Kafka消費(fèi)實(shí)時(shí)數(shù)據(jù)做窗口聚合另一個(gè)Spark SQL批處理作業(yè)直接讀取CSV或HBase歷史數(shù)據(jù)做離線統(tǒng)計(jì)兩個(gè)作業(yè)共用同一套清洗函數(shù)保證口徑一致。存儲(chǔ)層包括HBase和MySQLHBase存客流明細(xì)和分鐘級(jí)聚合結(jié)果MySQL存站點(diǎn)線路表。應(yīng)用層提供HTTP接口和簡(jiǎn)單前端頁面把統(tǒng)計(jì)結(jié)果和預(yù)測(cè)曲線展示出來項(xiàng)目里的HTTP接口調(diào)試文件就對(duì)應(yīng)這一層。整個(gè)鏈路的拓?fù)鋵?duì)應(yīng)關(guān)系用一個(gè)bash代碼塊描述方便照著搭環(huán)境# 數(shù)據(jù)鏈路采集 - 緩沖 - 計(jì)算 - 存儲(chǔ) - 展示 # Nginx日志(模擬閘機(jī)請(qǐng)求) # └─ Logstash(file input - grok過濾 - kafka output) # └─ Kafka topic: metro-flows # ├─ Spark Streaming(窗口聚合) - HBase # └─ Spark SQL(離線清洗統(tǒng)計(jì)) - HBase # └─ 后端HTTP接口 - 前端圖表展示每一層都是黑匣子上一層的輸出格式就是下一層的輸入約定。我在設(shè)計(jì)時(shí)強(qiáng)制約定了一條數(shù)據(jù)規(guī)范所有寫入Kafka的消息統(tǒng)一為JSON格式包含timestamp、stationId、lineId、direction、count五個(gè)字段。這個(gè)約定是整個(gè)項(xiàng)目少踩坑的關(guān)鍵后面所有模塊都圍繞這五個(gè)字段編程避免各層各寫一套解析邏輯。3.2 工程文件結(jié)構(gòu)畢業(yè)設(shè)計(jì)壓縮包里每個(gè)文件是干什么的拿到一份完整畢設(shè)源碼第一件事是看懂文件結(jié)構(gòu)。這套項(xiàng)目的壓縮包結(jié)構(gòu)很典型各文件職責(zé)如下表文件/目錄作用hbase.commandHBase建表和預(yù)分區(qū)操作命令腳本logstash-nginx.configLogstash采集Nginx日志的配置文件szmc.net-metro.csv地鐵客流模擬測(cè)試數(shù)據(jù)search.httpHTTP接口調(diào)試文件演示查詢接口szt-api.http后端API接口調(diào)用示例.editorconfig / .gitignore編碼規(guī)范和Git忽略規(guī)則hbase.command這一步會(huì)預(yù)先創(chuàng)建客流表并設(shè)置預(yù)分區(qū)避免數(shù)據(jù)量上來之后Region熱點(diǎn)。logstash-nginx.config是采集層的核心配置。而search.http和szt-api.http則保存了接口請(qǐng)求示例演示時(shí)直接點(diǎn)擊發(fā)送就能看到返回結(jié)果不需要額外寫測(cè)試工具。這些文件共同構(gòu)成了一個(gè)能演示、能擴(kuò)展、能寫進(jìn)文檔的完整工程。3.3 數(shù)據(jù)庫設(shè)計(jì)HBase行鍵怎么設(shè)計(jì)才不踩熱點(diǎn)客流表的設(shè)計(jì)是整個(gè)存儲(chǔ)層的關(guān)鍵。HBase表名設(shè)計(jì)為metro_flow行鍵采用“站點(diǎn)ID反轉(zhuǎn) 時(shí)間戳反轉(zhuǎn)”的組合方式。為什么要反轉(zhuǎn)如果直接用自然時(shí)間戳作為行鍵前綴所有寫入都會(huì)打到同一個(gè)Region上形成寫熱點(diǎn)。將時(shí)間戳反轉(zhuǎn)后新數(shù)據(jù)分散到不同Region寫入壓力被攤開同時(shí)保留了一定程度的時(shí)間順序掃描能力。建表和預(yù)分區(qū)命令如下# hbase.command 內(nèi)容示例創(chuàng)建客流表并預(yù)分區(qū) create metro_flow, info, {NUMREGIONS 8, SPLITALGO HexStringSplit} # 行鍵示例: # stationId反轉(zhuǎn)(例如S001001反轉(zhuǎn)) - timestamp反轉(zhuǎn)(例如20240509103000反轉(zhuǎn)) # put時(shí)直接指定行鍵寫入 put metro_flow, 100100S-00003010504202, info:direction, 1 put metro_flow, 100100S-00003010504202, info:passenger, 8行鍵里stationId在前是為了讓同一站點(diǎn)的數(shù)據(jù)盡量落在相鄰Region方便按站點(diǎn)掃描時(shí)間戳反轉(zhuǎn)在后則避免了順序?qū)憻狳c(diǎn)。列族設(shè)計(jì)只有一個(gè)info列字段包括direction、passenger、lineId等。實(shí)際開發(fā)中建議加上“日期”作為行鍵前綴的一部分比如“20240509-100100S-00003010504202”這樣按天掃描數(shù)據(jù)時(shí)直接指定前綴效率更高。4. 核心模塊實(shí)現(xiàn)客流清洗、離線統(tǒng)計(jì)與實(shí)時(shí)預(yù)測(cè)的代碼落地4.1 數(shù)據(jù)接入與解析Logstash如何從Nginx日志提取客流字段Logstash配置是數(shù)據(jù)進(jìn)入計(jì)算層的第一道關(guān)口。它監(jiān)聽Nginx的access.log用grok插件把日志行拆成結(jié)構(gòu)化字段再通過date插件統(tǒng)一時(shí)間戳格式最后輸出到Kafka。配置文件的關(guān)鍵部分如下# logstash-nginx.config 核心片段 input { file { path /var/log/nginx/access.log start_position beginning } } filter { grok { match { message %{IP:client_ip} - - \[%{HTTPDATE:ts}\] \%{WORD:method} %{URIPATH:uri} HTTP/%{NUMBER:httpv}\ %{NUMBER:resp} %{NUMBER:bytes} \%{DATA:referrer}\ \%{DATA:agent}\ } } date { match [ ts, dd/MMM/yyyy:HH:mm:ss Z ] target timestamp } mutate { add_field { stationId %{uripath2stationid} } } } output { kafka { bootstrap_servers localhost:9092 topic_id metro-flows } }grok的match規(guī)則把Nginx默認(rèn)日志格式拆成結(jié)構(gòu)化字段date插件負(fù)責(zé)把日志里的時(shí)間字符串轉(zhuǎn)成Logstash內(nèi)部的標(biāo)準(zhǔn)時(shí)間戳這一步如果漏掉后面Spark按時(shí)間窗口聚合時(shí)會(huì)出現(xiàn)時(shí)間錯(cuò)亂。add_field是運(yùn)行時(shí)動(dòng)態(tài)補(bǔ)充字段比如從URI里解析出站點(diǎn)ID生產(chǎn)上通常用uri里的查詢參數(shù)或路徑片段作為站點(diǎn)標(biāo)識(shí)。寫配置時(shí)最容易出問題的是日志格式與grok表達(dá)式不匹配。常見做法是先抓一條真實(shí)日志用grok debugger在線驗(yàn)證正則確認(rèn)所有字段都提取成功后再把配置寫入文件。項(xiàng)目文件里這份logstash-nginx.config就是按這個(gè)思路調(diào)通的。4.2 實(shí)時(shí)客流統(tǒng)計(jì)Spark Streaming按分鐘窗口聚合進(jìn)出站流量實(shí)時(shí)模塊的目標(biāo)是統(tǒng)計(jì)每個(gè)站點(diǎn)最近5分鐘的出入口進(jìn)出站人數(shù)。這里用Spark Streaming的reduceByKeyAndWindow算子按滑動(dòng)窗口做增量聚合。核心代碼如下from pyspark.streaming import StreamingContext from pyspark.streaming.kafka import KafkaUtils import json ssc StreamingContext(spark.sparkContext, batch_duration10) kafka_params {bootstrap.servers: localhost:9092, group.id: metro-stream} stream KafkaUtils.createDirectStream( ssc, [metro-flows], kafka_params, valueDecoderlambda x: json.loads(x.decode(utf-8)) ) def parse_and_key(msg): rec msg[1] # msg結(jié)構(gòu): (topic, value) return (rec[stationId], rec[direction]), rec[passenger] counts (stream.map(parse_and_key) .reduceByKeyAndWindow(lambda a, b: a b, lambda a, b: a - b, window_seconds300, slide_seconds60)) def save_to_hbase(rdd): # 將聚合結(jié)果寫入HBase行鍵按 日期-站點(diǎn)-窗口起始時(shí)間 構(gòu)造 rdd.foreachPartition(write_batch) counts.foreachRDD(save_to_hbase) ssc.start() ssc.awaitTermination()這段代碼里batch_duration是10秒Spark每隔10秒處理一批Kafka消息。reduceByKeyAndWindow的window_seconds設(shè)為300秒代表統(tǒng)計(jì)最近5分鐘slide_seconds設(shè)為60秒代表結(jié)果每分鐘更新一次。增量計(jì)算的巧妙之處在于它只對(duì)新到的數(shù)據(jù)和滑出的數(shù)據(jù)做加減不用每60秒全量重算這在窗口較長(zhǎng)時(shí)能明顯節(jié)省計(jì)算開銷。save_to_hbase函數(shù)需要注意批量提交不要每條記錄單獨(dú)put否則HBase服務(wù)端壓力很大本地跑不明顯集群上跑容易超時(shí)。實(shí)際項(xiàng)目里通常累積一批用HBase的BufferedMutator批量寫入。4.3 離線分析與預(yù)測(cè)Spark SQL統(tǒng)計(jì)時(shí)段客流MLlib訓(xùn)練預(yù)測(cè)模型離線分析解決“按小時(shí)、按站點(diǎn)看客流走勢(shì)”。Spark SQL讀取CSV后注冊(cè)臨時(shí)表寫標(biāo)準(zhǔn)SQL完成統(tǒng)計(jì)// 離線統(tǒng)計(jì)讀取CSV為DataFrame并注冊(cè)臨時(shí)表 DatasetRow flows spark.read() .option(header, true) .option(inferSchema, true) .csv(hdfs:///data/szmc.net-metro.csv); flows.createOrReplaceTempView(metro_flows); // 按站點(diǎn)和小時(shí)統(tǒng)計(jì)進(jìn)出站客流 spark.sql( SELECT stationId, hour(ts) as hour, direction, SUM(passenger) as total FROM metro_flows GROUP BY stationId, hour(ts), direction ORDER BY stationId, hour) .write().mode(overwrite).saveAsTable(result_hourly_flow);這里有兩個(gè)細(xì)節(jié)容易忽略。第一CSV的header字段名要保持與SQL引用名一致否則報(bào)unknown column第二hour(ts)要求ts列已經(jīng)是時(shí)間戳類型如果CSV里是字符串讀入后要先用to_timestamp轉(zhuǎn)換再注冊(cè)臨時(shí)表。預(yù)測(cè)模塊使用MLlib的線性回歸。特征選取同一站點(diǎn)前一周同一小時(shí)的客流、當(dāng)天是否為工作日、前一天同時(shí)段客流三個(gè)維度標(biāo)簽是當(dāng)前小時(shí)的客流。訓(xùn)練集和測(cè)試集按7比3劃分代碼如下from pyspark.ml.regression import LinearRegression from pyspark.ml.feature import VectorAssembler feature_cols [last_week_same_hour, is_workday, yesterday_same_hour] assembler VectorAssembler(inputColsfeature_cols, outputColfeatures) data assembler.transform(feature_df) train, test data.randomSplit([0.7, 0.3], seed42) lr LinearRegression(featuresColfeatures, labelColpassenger, maxIter20) model lr.fit(train) test_result model.evaluate(test) # 輸出預(yù)測(cè)誤差指標(biāo)RMSE 和 R2 print(RMSE:, test_result.rootMeanSquaredError) print(R2:, test_result.r2)線性回歸的模型解釋性強(qiáng)答辯時(shí)能直接講清楚每個(gè)特征對(duì)客流的影響權(quán)重這一點(diǎn)比黑匣子式的隨機(jī)森林或深度學(xué)習(xí)模型更受導(dǎo)師認(rèn)可。5. 高頻踩坑點(diǎn)與排查方案版本兼容、內(nèi)存配置與數(shù)據(jù)傾斜5.1 現(xiàn)象Spark作業(yè)頻繁O(jiān)OM做實(shí)時(shí)窗口聚合時(shí)作業(yè)運(yùn)行十幾分鐘后突然報(bào)Executor Lost和Container killed。表面是內(nèi)存不足實(shí)際是executor內(nèi)存設(shè)置與窗口計(jì)算量不匹配。reduceByKeyAndWindow需要將整個(gè)窗口的數(shù)據(jù)保留在內(nèi)存中做增量計(jì)算窗口越大內(nèi)存占用越高。解決方法是先估算單批次數(shù)據(jù)量和窗口內(nèi)批次數(shù)量再設(shè)置executor內(nèi)存同時(shí)把spark.streaming.kafka.maxRatePerPartition調(diào)低限制每秒拉取上限給計(jì)算留出余量。我一般會(huì)按“executor內(nèi)存 單批次預(yù)估大小 × 窗口內(nèi)批次數(shù) / 并行度 × 3”來粗略估算。5.2 現(xiàn)象HBase寫入超時(shí)RegionServer頻繁宕機(jī)批量put時(shí)某幾個(gè)RegionServer負(fù)載極高因?yàn)樾墟I設(shè)計(jì)沒做反轉(zhuǎn)時(shí)間戳順序?qū)懭雽?dǎo)致所有新數(shù)據(jù)集中在一個(gè)Region。解決方法是按前面第3.3節(jié)的方案將時(shí)間戳反轉(zhuǎn)作為行鍵后綴并對(duì)表做預(yù)分區(qū)如果嫌反轉(zhuǎn)麻煩也可以對(duì)行鍵做加鹽處理在行鍵前拼一位隨機(jī)數(shù)或取模值讓數(shù)據(jù)均勻分布到多個(gè)Region。5.3 現(xiàn)象Logstash解析出來的時(shí)間比實(shí)際時(shí)間晚8小時(shí)Nginx日志里的時(shí)間是本地時(shí)間date插件解析后默認(rèn)按UTC處理導(dǎo)致Kafka里的時(shí)間戳整體偏移8小時(shí)Spark窗口統(tǒng)計(jì)的時(shí)間段全部錯(cuò)位。這個(gè)坑在本地測(cè)試時(shí)不容易發(fā)現(xiàn)等到比對(duì)實(shí)時(shí)曲線和真實(shí)客流時(shí)段才對(duì)不上。解決方式是在date插件里指定timezone為“Asia/Shanghai”或者讓寫入Kafka前統(tǒng)一轉(zhuǎn)換為UTCSpark消費(fèi)時(shí)再轉(zhuǎn)換回本地時(shí)間。整套鏈路必須統(tǒng)一時(shí)區(qū)口徑最忌諱各層各轉(zhuǎn)一次。5.4 現(xiàn)象Kafka消費(fèi)者lag只增不減實(shí)時(shí)作業(yè)處理速度跟不上數(shù)據(jù)產(chǎn)生速度運(yùn)行一段時(shí)間后堆積嚴(yán)重。用Kafka自帶的消費(fèi)組命令查看lag持續(xù)增長(zhǎng)說明單批次處理時(shí)間已經(jīng)超過batch_duration。排查先看是不是某個(gè)task卡在HBase寫入上其次看是否有數(shù)據(jù)傾斜導(dǎo)致部分task處理量大。解決方向有三個(gè)增加executor并行度、降低maxRatePerPartition限制、優(yōu)化HBase寫入為批量異步提交。優(yōu)先調(diào)并行度因?yàn)樗苤苯涌s短批處理時(shí)間。5.5 現(xiàn)象本地能跑通提交到集群就報(bào)文件不存在本地開發(fā)用相對(duì)路徑讀CSV提交到集群后當(dāng)前目錄不再是本地目錄導(dǎo)致FileNotFound。排查時(shí)先確認(rèn)代碼里所有路徑都寫成HDFS路徑或絕對(duì)路徑再檢查Spark作業(yè)提交時(shí)有沒有把依賴jar包用--jars帶上。這個(gè)坑屬于環(huán)境差異問題通常在代碼評(píng)審階段就能發(fā)現(xiàn)我自己的方法是提交前過一遍清單所有路徑不帶相對(duì)路徑、所有外部依賴已打包、所有端口已釋放。6. 驗(yàn)收與答辯用測(cè)試數(shù)據(jù)結(jié)果證明系統(tǒng)可用6.1 功能驗(yàn)收三個(gè)場(chǎng)景的輸入、輸出與判斷標(biāo)準(zhǔn)答辯前必須能現(xiàn)場(chǎng)演示功能驗(yàn)收表要提前準(zhǔn)備好每項(xiàng)能用測(cè)試數(shù)據(jù)當(dāng)場(chǎng)跑通。驗(yàn)收?qǐng)鼍拜斎腩A(yù)期輸出判定標(biāo)準(zhǔn)離線時(shí)段統(tǒng)計(jì)szmc.net-metro.csv每小時(shí)每站點(diǎn)進(jìn)出站人數(shù)與手工抽樣統(tǒng)計(jì)誤差1%實(shí)時(shí)窗口聚合回放假客流數(shù)據(jù)流最近5分鐘各站客流延遲不超過1個(gè)窗口周期客流預(yù)測(cè)歷史30天客流序列下一小時(shí)預(yù)測(cè)值RMSE在可接受范圍趨勢(shì)跟隨正確離線統(tǒng)計(jì)的驗(yàn)證方式是抽幾行CSV數(shù)據(jù)手動(dòng)加和與SQL統(tǒng)計(jì)結(jié)果對(duì)比實(shí)時(shí)聚合的驗(yàn)證方式是在回放腳本中人為插入一個(gè)高峰流量尖峰看5分鐘統(tǒng)計(jì)是否同步突起預(yù)測(cè)模塊則準(zhǔn)備一條“工作日早高峰”和一條“周末平峰”的對(duì)比曲線直觀展示模型能區(qū)分不同日期類型。6.2 答辯演示順序與講解節(jié)奏答辯演示不要從代碼開始講而是從數(shù)據(jù)鏈路講起。先展示HBase里的原始表結(jié)構(gòu)再打開Logstash配置說明采集方式接著跑一條離線統(tǒng)計(jì)SQL并把結(jié)果表和柱狀圖展示出來最后切到預(yù)測(cè)模塊的曲線對(duì)比圖。每講完一個(gè)模塊主動(dòng)拋出一個(gè)自己踩過的坑和解決方案這比背概念更能體現(xiàn)項(xiàng)目真實(shí)性。演示用的圖表建議提前用Python或前端圖表庫生成好不要現(xiàn)場(chǎng)等Spark作業(yè)跑。線上環(huán)境只保留一個(gè)實(shí)時(shí)統(tǒng)計(jì)任務(wù)作為動(dòng)態(tài)演示其余結(jié)果做成靜態(tài)圖或預(yù)計(jì)算緩存避免現(xiàn)場(chǎng)網(wǎng)絡(luò)波動(dòng)導(dǎo)致作業(yè)失敗。6.3 收尾按理說以上流程走一遍系統(tǒng)能不能交付已經(jīng)心里有數(shù)。但我?guī)н^的項(xiàng)目中還是有人疏漏了行鍵設(shè)計(jì)或時(shí)區(qū)問題答辯現(xiàn)場(chǎng)才暴露。從那以后我每次提交畢設(shè)或項(xiàng)目方案前都會(huì)強(qiáng)制自己走一遍清單數(shù)據(jù)管道每一層單獨(dú)驗(yàn)證、HBase預(yù)分區(qū)已設(shè)置、時(shí)區(qū)口徑統(tǒng)一、所有路徑和依賴已核對(duì)。這套基于Spark的地鐵大數(shù)據(jù)客流分析系統(tǒng)只要按第2章到第6章的路徑執(zhí)行是能在有限時(shí)間里穩(wěn)定跑通的。希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取