這兩年被問得最多的一個(gè)問題不是“Hadoop怎么學(xué)”而是“我們那一批傳感器每天上報(bào)幾百億條數(shù)據(jù)寫入沒問題但想查點(diǎn)什么一個(gè)查詢跑半小時(shí)怎么辦”。物聯(lián)網(wǎng)的傳感器數(shù)據(jù)跟傳統(tǒng)互聯(lián)網(wǎng)日志完全不是一回事——設(shè)備數(shù)量動(dòng)輒百萬級(jí)每臺(tái)設(shè)備幾秒一條數(shù)據(jù)一天下來就是幾百億條記錄而且數(shù)據(jù)格式五花八門時(shí)序特征極強(qiáng)還有大量噪音和缺失。很多人一開始用傳統(tǒng)數(shù)據(jù)庫扛扛到幾千萬條就明顯卡頓換時(shí)序數(shù)據(jù)庫能解決一部分但涉及復(fù)雜分析、歷史歸檔、跟業(yè)務(wù)數(shù)據(jù)做關(guān)聯(lián)又力不從心。這時(shí)候Hadoop生態(tài)的價(jià)值就體現(xiàn)出來了——它不是為了存數(shù)據(jù)而存數(shù)據(jù)而是為了讓你在幾十億條傳感器記錄上還能跑出個(gè)結(jié)果來。這篇文章我從實(shí)際項(xiàng)目的角度把“Hadoop物聯(lián)網(wǎng)傳感器數(shù)據(jù)”這個(gè)組合拆開講數(shù)據(jù)鏈路怎么搭、清洗策略怎么做、存儲(chǔ)模型怎么建、跑分析時(shí)有哪些坑最后用一個(gè)典型場(chǎng)景復(fù)盤收尾。適合正在做物聯(lián)網(wǎng)平臺(tái)、準(zhǔn)備上Hadoop處理設(shè)備數(shù)據(jù)的團(tuán)隊(duì)也適合畢業(yè)設(shè)計(jì)選了物聯(lián)網(wǎng)方向的同學(xué)們參考。1. 物聯(lián)網(wǎng)傳感器數(shù)據(jù)的四種“脾氣”為什么非Hadoop不可搞過物聯(lián)網(wǎng)的人都有體會(huì)傳感器數(shù)據(jù)跟人工錄入的“業(yè)務(wù)數(shù)據(jù)”完全兩個(gè)物種。我經(jīng)常打比方傳感器數(shù)據(jù)像是流水線上的零件——每個(gè)單獨(dú)看起來都差不多數(shù)量卻大到嚇人而業(yè)務(wù)數(shù)據(jù)像是檔案室里的文件——數(shù)量少但每份都很重要格式也要精雕細(xì)琢。拿處理文件的方式去處理零件必然出問題。1.1 高頻寫入幾秒鐘一條量級(jí)跟日志沒法比傳統(tǒng)互聯(lián)網(wǎng)日志寫入峰值一臺(tái)服務(wù)器每秒鐘幾百條就算高了但一臺(tái)工業(yè)網(wǎng)關(guān)后面掛了上百個(gè)傳感器每個(gè)傳感器3秒上報(bào)一次這個(gè)網(wǎng)關(guān)每秒就有幾十條數(shù)據(jù)。一個(gè)中型工廠幾百臺(tái)網(wǎng)關(guān)就是每秒上萬條寫入。這還不算共享單車、智慧路燈、環(huán)境監(jiān)測(cè)這類全國性場(chǎng)景。我見過一個(gè)項(xiàng)目40萬個(gè)設(shè)備每2秒一條心跳數(shù)據(jù)光是一天的數(shù)據(jù)量就是17億條2TB壓縮后。這是物聯(lián)網(wǎng)數(shù)據(jù)跟普通日志最本質(zhì)的區(qū)別——持續(xù)不斷從不睡覺全年無休。這種高頻寫入場(chǎng)景下傳統(tǒng)關(guān)系型數(shù)據(jù)庫的瓶頸很明顯每一行插入都要走索引、走日志寫入吞吐上不去。Hadoop生態(tài)里的HDFS解決了存儲(chǔ)層的大規(guī)模吞吐問題靠的是大塊順序?qū)懚鳮afka這種消息中間件則解決了“接入層削峰”的問題——后面細(xì)說。1.2 時(shí)序特征極強(qiáng)每條數(shù)據(jù)都帶著時(shí)間戳但時(shí)間戳最不可信傳感器數(shù)據(jù)本質(zhì)上是時(shí)序數(shù)據(jù)一個(gè)設(shè)備ID 一個(gè)時(shí)間戳 一個(gè)或多個(gè)測(cè)量值。這個(gè)特性決定了存儲(chǔ)模型可以高度簡(jiǎn)化——按設(shè)備分桶按時(shí)間排序所有查詢要么是按設(shè)備查一段歷史要么是按時(shí)間范圍跨設(shè)備掃描。跟業(yè)務(wù)數(shù)據(jù)那種多表關(guān)聯(lián)的復(fù)雜結(jié)構(gòu)比起來傳感器數(shù)據(jù)的模型簡(jiǎn)單得多但數(shù)據(jù)量卻大幾個(gè)量級(jí)。但這里有個(gè)反直覺的坑時(shí)間戳看起來是數(shù)據(jù)自帶的屬性實(shí)際上恰恰是傳感器數(shù)據(jù)里最不靠譜的字段。設(shè)備時(shí)鐘漂移、網(wǎng)關(guān)緩存重傳、網(wǎng)絡(luò)延遲都會(huì)讓數(shù)據(jù)到達(dá)時(shí)間和數(shù)據(jù)產(chǎn)生時(shí)間出現(xiàn)偏差有的甚至差幾個(gè)小時(shí)。我在后面有一節(jié)專門講這個(gè)問題這里先提個(gè)醒——如果你把入庫時(shí)間當(dāng)成了傳感器產(chǎn)生時(shí)間那后面的分析結(jié)果會(huì)很離譜。1.3 格式參差不齊幾十種設(shè)備幾十種協(xié)議聚在一起就是爛攤子一個(gè)項(xiàng)目里很少只有一種傳感器。溫度傳感器、濕度傳感器、振動(dòng)傳感器、能耗表計(jì)、定位追蹤器……每種的報(bào)文格式都不一樣。有的上報(bào)字段叫temp有的叫temperature有的干脆是data: {v: 23.5}這種嵌在JSON里的。更過分的是不同批次的固件版本字段含義還會(huì)變。這不是代碼規(guī)范問題而是設(shè)備廠商太多、協(xié)議標(biāo)準(zhǔn)跟不上的現(xiàn)實(shí)。這部分臟活累活在Hadoop架構(gòu)里通常拆成兩層解決接入層做“格式歸一化”把亂七八糟的報(bào)文解析成統(tǒng)一的JSON或Avro格式分析層再做“字段標(biāo)準(zhǔn)化”把歷史數(shù)據(jù)統(tǒng)一到一個(gè)口徑。這也是為什么我在下一節(jié)強(qiáng)調(diào)Kafka的schema管理能力。1.4 數(shù)據(jù)的價(jià)值密度低但分析需求卻很高單條傳感器數(shù)據(jù)的價(jià)值密度極低——“設(shè)備A在10:03:27時(shí)刻的溫度是26.1℃”——這條信息基本沒用。但成百上千設(shè)備的時(shí)間序列拼在一起就能看出設(shè)備是否異常、產(chǎn)線是否過載、能源消耗是否異常。也就是說物聯(lián)網(wǎng)數(shù)據(jù)的特點(diǎn)是單體沒用聚合才有價(jià)值。這個(gè)特性決定了存儲(chǔ)不能丟但又不能粒度太細(xì)地長期全保留分析既要能全量掃描比如找出上個(gè)月所有設(shè)備的溫升曲線又要能快速定位比如查某個(gè)設(shè)備3小時(shí)前的瞬時(shí)值。Hadoop生態(tài)可以同時(shí)滿足這兩類需求HDFS/HBase管存儲(chǔ)Hive/Spark管全量分析HBase或Redis管點(diǎn)查加速。這也是為什么我不建議只上一個(gè)時(shí)序數(shù)據(jù)庫——時(shí)序庫確實(shí)在寫入和點(diǎn)查上有優(yōu)勢(shì)但跨設(shè)備復(fù)雜聚合分析、跟其他系統(tǒng)的數(shù)據(jù)做關(guān)聯(lián)還是Hadoop生態(tài)更順手。2. 整條數(shù)據(jù)鏈路怎么搭傳感器、網(wǎng)關(guān)、Kafka到HDFS先給一個(gè)我自己慣用的參考架構(gòu)再挨個(gè)拆解每一層為什么要這么選。傳感器設(shè)備 → 邊緣網(wǎng)關(guān) → Kafka數(shù)據(jù)接入層→ 流處理/清洗 → HDFS/HBase數(shù)據(jù)存儲(chǔ)層→ Hive/Spark分析層→ 應(yīng)用2.1 為什么中間非要加一層Kafka入湖和入庫是兩件事很多第一次做物聯(lián)網(wǎng)數(shù)據(jù)平臺(tái)的同學(xué)會(huì)問傳感器數(shù)據(jù)直接寫到HDFS不就行了干嘛要在中間加一個(gè)Kafka答案是——HDFS適合“批量落盤”不適合“每秒鐘幾萬條實(shí)時(shí)寫入”。HDFS的優(yōu)勢(shì)是大塊順序?qū)憽⒏咄掏屡繉?dǎo)入每來一條就寫入一次會(huì)產(chǎn)生大量小文件后面專門講。而Kafka的作用就是緩沖和削峰傳感器數(shù)據(jù)先沖到Kafka里下游不管是用Flume還是用Spark Streaming按自己的節(jié)奏批量寫入HDFS。這樣做還有另一層好處數(shù)據(jù)入湖和數(shù)據(jù)處理解耦了。設(shè)備不用關(guān)心下游存儲(chǔ)系統(tǒng)的死活Kafka里的數(shù)據(jù)可以先攢著哪怕下游HDFS集群重啟、跑批任務(wù)掛了數(shù)據(jù)一條不丟。Kafka默認(rèn)保留策略是7天這7天就是你的“后悔藥窗口”和“追數(shù)窗口”。2.2 選型對(duì)比Flume還是Kafka Connector有了Kafka之后從Kafka到HDFS這一段有三個(gè)方案經(jīng)常被拿來比Flume、Kafka Connect特別是HDFS Sink Connector、以及直接用Spark Streaming寫。我列個(gè)表用實(shí)際項(xiàng)目經(jīng)驗(yàn)說話方案優(yōu)點(diǎn)缺點(diǎn)適合場(chǎng)景Flume Kafka Source穩(wěn)定上手快整套架構(gòu)都是Apache系配置繁瑣自定義Interceptor要寫Java監(jiān)控能力一般日志型數(shù)據(jù)、簡(jiǎn)單管道Kafka Connect HDFS Sink連接器生態(tài)好支持Avro/Parquet/ORC自動(dòng)分區(qū)有schema管理依賴Schema Registry版本匹配坑多兼容性問題在CDH/HDP之間尤其明顯標(biāo)準(zhǔn)化程度高、字段變更少的管道Spark Structured Streaming一步到位邊寫邊清洗結(jié)局大白于天下寫完每批數(shù)據(jù)即可做處理處理邏輯寫不好容易拖垮寫入性能資源消耗比前兩個(gè)高既要做清洗又要寫庫管道邏輯復(fù)雜我自己的偏好是管道簡(jiǎn)單就用Flume管道邏輯復(fù)雜就直接Spark Structured StreamingKafka Connect反而用得少——因?yàn)樗押芏嗵幚磉壿嬒拗圃诹伺渲脤右坏┯龅綐I(yè)務(wù)字段映射這類需求配置比寫代碼還痛苦。不過這是個(gè)人喜好團(tuán)隊(duì)技術(shù)棧不同選擇不同有一點(diǎn)是公認(rèn)的從Kafka到HDFS的這層管道一定要支持按批提交、失敗重試和流量監(jiān)控缺一個(gè)后面都會(huì)很被動(dòng)。2.3 存儲(chǔ)層選擇HDFS還是HBase兩條腿走路物聯(lián)網(wǎng)傳感器數(shù)據(jù)的存儲(chǔ)我建議兩條腿走路一份放HDFS一份放HBase或者用其他列式存儲(chǔ)。HDFS Parquet/ORC用于歷史歸檔和批量分析。按時(shí)間分區(qū)存儲(chǔ)保留周期可以很長一年甚至幾年。分析任務(wù)是讀這類數(shù)據(jù)。HBase用于最近N天的點(diǎn)查和實(shí)時(shí)查詢。比如“查某個(gè)設(shè)備當(dāng)前狀態(tài)”、“查某個(gè)設(shè)備最近一小時(shí)曲線”走HBase的RowKey索引非?？臁owKey設(shè)計(jì)一般是設(shè)備ID逆序 時(shí)間戳避免熱點(diǎn)后面細(xì)說。如果你不想引入HBase也可以用HDFS Hive直接扛點(diǎn)查——建好分區(qū)表用Spark SQL按分區(qū)過濾查千萬級(jí)設(shè)備量下響應(yīng)一般在秒級(jí)到十秒級(jí)很多場(chǎng)景夠用了。非要說什么時(shí)候必須上HBase那只有“查詢要求毫秒到百毫秒級(jí)”的場(chǎng)景比如實(shí)時(shí)告警聯(lián)動(dòng)、大屏點(diǎn)查。3. 讓數(shù)據(jù)“干凈”地入庫傳感器數(shù)據(jù)的清洗策略與實(shí)操“臟數(shù)據(jù)進(jìn)庫分析結(jié)果就是垃圾?！边@句廢話在物聯(lián)網(wǎng)領(lǐng)域尤其重要——因?yàn)檫@行業(yè)的數(shù)據(jù)臟法跟別處不一樣不是人為錄入錯(cuò)誤而是設(shè)備層面的物理性失真。采集環(huán)節(jié)不可控所以清洗策略必須在入庫前做扎實(shí)。3.1 三類最常見的傳感器“臟數(shù)據(jù)”及判據(jù)我歸納下來傳感器數(shù)據(jù)清洗主要解決三類問題1重復(fù)數(shù)據(jù)同一時(shí)刻同一設(shè)備上報(bào)了兩條一模一樣的記錄。原因一般是設(shè)備的重傳機(jī)制網(wǎng)絡(luò)抖動(dòng)導(dǎo)致ACK沒到設(shè)備重發(fā)或者網(wǎng)關(guān)轉(zhuǎn)發(fā)了兩次。判據(jù)就是設(shè)備ID 來源時(shí)間戳這兩列組合去重。特別注意去重不能只看JSON是否完全一樣因?yàn)閮纱沃貍鞯臄?shù)據(jù)可能部分字段不同比如接收時(shí)間不同要用業(yè)務(wù)主鍵去重。2離群值/超范圍值溫度傳感器報(bào)了500℃濕度報(bào)了-20%這種數(shù)據(jù)明顯不物理。判定方法是給每個(gè)測(cè)點(diǎn)配置合理的上下限。但這里有個(gè)容易踩的坑——不同應(yīng)用場(chǎng)景的閾值不一樣。同一塊溫度傳感器用在常溫廠房0~40℃和用在冷鏈-30~10℃上正常范圍完全不一樣。所以閾值配置必須跟著設(shè)備類型走不能寫死在處理代碼里。3缺失數(shù)據(jù)與虛假數(shù)據(jù)設(shè)備掉線會(huì)導(dǎo)致一段時(shí)間完全沒有數(shù)據(jù)設(shè)備故障則可能反復(fù)上報(bào)同一個(gè)值比如一直報(bào)24.0。缺失數(shù)據(jù)還好辦補(bǔ)一個(gè)空或標(biāo)記缺失即可虛假數(shù)據(jù)最難搞要結(jié)合“數(shù)值隨時(shí)間是否變化”來判定。我在項(xiàng)目里用過最簡(jiǎn)單的判據(jù)同一測(cè)點(diǎn)連續(xù)10條數(shù)據(jù)值完全一樣就標(biāo)記為“疑似異常值”寫入清洗表里人工抽檢。3.2 清洗在哪個(gè)環(huán)節(jié)做端側(cè)、接入層、還是分析層三處都有活干但職責(zé)不同端側(cè)邊緣網(wǎng)關(guān)做格式解析、協(xié)議轉(zhuǎn)換、基礎(chǔ)校驗(yàn)必填字段是否缺失、報(bào)文是否合法。這里不做復(fù)雜邏輯因?yàn)榫W(wǎng)關(guān)算力有限而且升級(jí)困難——盡量少給端側(cè)加戲。接入層清洗任務(wù)/流處理做去重、時(shí)間口徑統(tǒng)一、字段標(biāo)準(zhǔn)化、值域校驗(yàn)。這一層是清洗的主戰(zhàn)場(chǎng)因?yàn)閿?shù)據(jù)到了這里才被集中看到可以做跨設(shè)備或按設(shè)備類型的規(guī)則判斷。分析層做深度清洗和異常檢測(cè)。比如后面要訓(xùn)練模型或做設(shè)備健康度評(píng)估這時(shí)才做滑動(dòng)窗口、趨勢(shì)判斷等復(fù)雜邏輯。一個(gè)具體的實(shí)操建議在接入層做“標(biāo)準(zhǔn)時(shí)間”字段。設(shè)備上報(bào)的device_time設(shè)備本地時(shí)間和receive_time網(wǎng)關(guān)/平臺(tái)接收時(shí)間都要保留但下游統(tǒng)一用event_time作為事件時(shí)間等于在清洗時(shí)就明確時(shí)間口徑。我在清洗任務(wù)里一般這樣規(guī)定優(yōu)先用設(shè)備本地時(shí)間但如果設(shè)備時(shí)鐘偏差跟接收時(shí)間比超過10分鐘則標(biāo)記為時(shí)鐘漂移數(shù)據(jù)改用接收時(shí)間并將原始時(shí)間放device_raw_time字段備查。后面講Spark Structured Streaming時(shí)再說watermark怎么配合這個(gè)口徑。3.3 清洗SQL長什么樣一段可以直接抄的示例假設(shè)清洗后統(tǒng)一輸出到Hive的ODS層表ods_sensor_data上游Kafka里的原始數(shù)據(jù)是JSON字符串我用Spark Structured Streaming做實(shí)時(shí)清洗核心邏輯如下省去環(huán)境初始化和參數(shù)配置只看清洗主體// Spark Structured Streaming 消費(fèi) Kafka清洗后寫入HDFS val raw spark .readStream .format(kafka) .option(kafka.bootstrap.servers, kfk01:9092,kfk02:9092) .option(subscribe, sensor_raw) .load() val parsed raw .selectExpr(CAST(value AS STRING) as json_str) .select(from_json($json_str, sensorSchema).as(data)) .select( $data.device_id.as(device_id), $data.temp.as(temp_raw), // 時(shí)間口徑統(tǒng)一設(shè)備時(shí)間優(yōu)先漂移則用接收時(shí)間 when( abs(unix_timestamp($data.sensor_time) - unix_timestamp($data.receive_time)) 600, $data.sensor_time ).otherwise($data.receive_time).cast(timestamp).as(event_time), // 值域校驗(yàn) when($data.temp.between(-40, 85), $data.temp).otherwise(lit(null)).as(temp), // 去重 $data.msg_id.as(dedup_key) ) // 以 msg_id 為主鍵做去重用stateful操作這一段可以直接做骨架去擴(kuò)展。有幾點(diǎn)值得說明from_json解析時(shí)一定要定義好sensorSchema字段變更時(shí)要做好兼容否則一個(gè)新設(shè)備類型上來就全管道崩了。去重用msg_id而不是設(shè)備ID時(shí)間戳拼接是因?yàn)椴煌?不同協(xié)議下msg_id的生成規(guī)則可能不同。最穩(wěn)妥的做法是設(shè)備ID傳感器時(shí)間戳隨機(jī)數(shù)三段拼一個(gè)唯一ID。對(duì)于斷言失敗的異常值我沒直接丟棄而是置NULL——這樣后面做分析時(shí)能區(qū)分“沒數(shù)據(jù)”和“數(shù)值不合法”兩種含義不同。4. 查得快才算數(shù)Hive分區(qū)建模與Spark分析實(shí)踐數(shù)據(jù)入庫只是第一步。真正的價(jià)值在“查”——但很多人發(fā)現(xiàn)數(shù)據(jù)是存進(jìn)去了Hive表也建了跑一個(gè)統(tǒng)計(jì)查詢要半小時(shí)Spark任務(wù)動(dòng)不動(dòng)OOM。這大概率是數(shù)據(jù)模型設(shè)計(jì)出了問。題傳感器數(shù)據(jù)的查詢模式高度固定設(shè)計(jì)好了90%的分析查詢都能走分區(qū)裁剪速度能差幾十倍。4.1 分區(qū)策略按時(shí)間分區(qū)還是按設(shè)備分組我的建議傳感器數(shù)據(jù)查詢有兩個(gè)天然維度設(shè)備維度和時(shí)間維度。Hive表怎么做分區(qū)直接決定了查詢效率。**按時(shí)間分區(qū)如按小時(shí)/天分區(qū)**是默認(rèn)方案絕大多數(shù)場(chǎng)景都適用。原因很簡(jiǎn)單傳感器數(shù)據(jù)分析里跨設(shè)備的時(shí)間范圍掃描是最常見的查詢——比如“查全天所有設(shè)備的平均溫度”、“查最近7天某型號(hào)設(shè)備的異常率”都是時(shí)間維度主導(dǎo)。按時(shí)間分區(qū)后這類查詢只需讀取對(duì)應(yīng)分區(qū)的數(shù)據(jù)掃描量從“全表”降到“一天的量”。但這里有幾個(gè)實(shí)操層面的建議分區(qū)粒度不要太小。監(jiān)控?cái)?shù)據(jù)量不大的場(chǎng)景按天分區(qū)夠了量太大再考慮按小時(shí)。我見過有人按5分鐘分區(qū)結(jié)果一個(gè)查詢要合并幾千個(gè)分區(qū)文件MapReduce的啟動(dòng)開銷比實(shí)際計(jì)算還大得不償失。分區(qū)列不要用dt這樣沒意義的字段干脆就叫event_date直接用清洗后的event_time來分區(qū)。這樣查詢時(shí)WHERE event_date 2024-06-01優(yōu)化器能精確裁剪。如果單體設(shè)備數(shù)據(jù)量極大比如一臺(tái)設(shè)備每天上千萬條記錄可以考慮“設(shè)備ID哈希分桶按時(shí)間分區(qū)”的雙層結(jié)構(gòu)。分桶字段是設(shè)備ID查詢某個(gè)設(shè)備的完整歷史時(shí)就能跳過大量不相關(guān)文件。但注意分桶數(shù)不能亂設(shè)要與文件大小匹配否則小文件問題會(huì)變本加厲。下面是建表模板可以直接抄CREATE TABLE dwd_sensor_data ( device_id STRING, device_type STRING, event_time TIMESTAMP, temp DOUBLE, humidity DOUBLE, vibration DOUBLE, ... ) PARTITIONED BY (event_date STRING) STORED AS PARQUET TBLPROPERTIES (parquet.compressionSNAPPY);4.2 列式存儲(chǔ)壓縮讓單條記錄再瘦一圈同樣的數(shù)據(jù)用TEXT存和用ParquetSnappy存查詢性能可以差5倍以上存儲(chǔ)空間可以壓縮60%以上。原理不復(fù)雜列式存儲(chǔ)只在讀取查詢涉及到的列時(shí)讀取對(duì)應(yīng)數(shù)據(jù)塊而傳感器數(shù)據(jù)一張表動(dòng)輒幾十個(gè)字段多數(shù)查詢只用其中兩三個(gè)字段——行式存儲(chǔ)要把一整行讀完才能拿到一個(gè)列的值。這就像你從一疊定制的紙質(zhì)表格里查所有人的手機(jī)號(hào)行式存儲(chǔ)要求翻完每一張完整表格列式存儲(chǔ)直接把“手機(jī)號(hào)”那一列抽出來。注意Parquet的另一個(gè)好處是內(nèi)置schema列名、列類型用Hive/Spark讀時(shí)不用再指定分隔符和字段順序少了不少解析錯(cuò)誤。我用的是Snappy壓縮——壓縮率比Gzip差一點(diǎn)但解壓速度快適合查詢頻繁的場(chǎng)景。冷數(shù)據(jù)想壓得更狠可以直接換ORCZlib但ORC在Spark里的支持沒Parquet那么順滑要看你的分析引擎主要用什么。4.3 Spark讀Hive數(shù)據(jù)兩個(gè)最常見的性能殺手講個(gè)真實(shí)數(shù)據(jù)一臺(tái)Spark任務(wù)讀1TB的Hive表做設(shè)備聚合分析第一次跑了一個(gè)多小時(shí)第二次五個(gè)小時(shí)第三次直接OOM。根因兩個(gè)殺手一讀出來的寬表。原始表有30個(gè)字段但分析只需要device_id、event_time、temp三個(gè)字段。代碼里如果有人用了SELECT *或者Spark的“謂詞下推”沒生效整表都被讀進(jìn)來了。解決辦法分析SQL里顯式寫出需要的列不要圖省事寫*檢查Spark物理計(jì)劃里PushedFilters是否生效。殺手二不合理的join策略。用Hive做設(shè)備基礎(chǔ)信息表和傳感器數(shù)據(jù)表的關(guān)聯(lián)如果基礎(chǔ)表只有幾萬行而傳感器表有幾十億行默認(rèn)的Shuffle Join會(huì)把幾十億行全部shuffle到所有節(jié)點(diǎn)傳輸量巨大。正確做法是廣播小表-- 使用Broadcast Join提示避免大表Shuffle SELECT /* BROADCAST(dim) */ s.device_id, d.region_name, AVG(s.temp) AS avg_temp FROM dwd_sensor_data s JOIN dim_device d ON s.device_id d.device_id WHERE s.event_date 2024-06-01 GROUP BY s.device_id, d.region_name;/* BROADCAST(dim) */這個(gè)提示能強(qiáng)制Spark把dim_device分發(fā)給每個(gè)Executor傳感器大表在本地完成關(guān)聯(lián)省掉一次幾億行的Shuffle。我見過很多團(tuán)隊(duì)優(yōu)化半天沒效果最后就是加了這個(gè)提示瞬間提升性能。4.4 實(shí)時(shí)分析怎么做Structured Streaming與“延遲數(shù)據(jù)”處理物聯(lián)網(wǎng)場(chǎng)景里實(shí)時(shí)和準(zhǔn)實(shí)時(shí)是一對(duì)繞不開的需求。要么是“設(shè)備數(shù)據(jù)延遲多久能看到”要么是“告警規(guī)則能不能在秒級(jí)觸發(fā)”。我的經(jīng)驗(yàn)是絕大部分物聯(lián)網(wǎng)場(chǎng)景不需要真正的毫秒級(jí)實(shí)時(shí)流處理秒級(jí)到分鐘級(jí)的準(zhǔn)實(shí)時(shí)就夠了。我也是這么落地的Kafka里取數(shù)據(jù)Spark Structured Streaming每30秒觸發(fā)一次micro batch做清洗和簡(jiǎn)單聚合后寫入結(jié)果表。關(guān)鍵在哪延遲數(shù)據(jù)。設(shè)備掉線一段時(shí)間后重新上線會(huì)把歷史緩存數(shù)據(jù)一股腦傳上來導(dǎo)致流任務(wù)里出現(xiàn)“昨天的事件今天才到達(dá)”。處理不好聚合結(jié)果會(huì)來回跳看板上的數(shù)字忽高忽低。解決方法是watermark水印機(jī)制——告訴流引擎“允許遲到多久”超時(shí)的一律丟棄或單獨(dú)走補(bǔ)償流程。我一般設(shè)10分鐘的watermark跟前面清洗時(shí)“設(shè)備時(shí)鐘漂移超過10分鐘改用接收時(shí)間”的口徑保持一致// 水印機(jī)制處理延遲數(shù)據(jù) events .withWatermark(event_time, 10 minutes) .groupBy(window($event_time, 1 minute), $device_id) .agg(avg($temp).as(avg_temp))5. 上線后才會(huì)遇到的三個(gè)經(jīng)典坑時(shí)鐘漂移、小文件與寫入熱點(diǎn)這一節(jié)寫的都是我在生產(chǎn)環(huán)境里真實(shí)踩過的坑踩一次抖三抖的那種。前兩個(gè)講了理論基礎(chǔ)這里專門講故障現(xiàn)場(chǎng)和修復(fù)過程。5.1 坑一設(shè)備時(shí)鐘漂移把“峰值分析”做成了“災(zāi)難現(xiàn)場(chǎng)”有個(gè)項(xiàng)目做工廠電力負(fù)荷分析目標(biāo)是看設(shè)備集群在“哪個(gè)時(shí)間段”用電最猛。上線兩周后BI團(tuán)隊(duì)反饋說數(shù)據(jù)完全沒法看——凌晨3點(diǎn)出現(xiàn)用電高峰白天反而波谷這跟工廠作息完全不符。排查過程是這樣的先看Kafka里的原始數(shù)據(jù)設(shè)備時(shí)間戳是正常的白天8點(diǎn)再查Hive表發(fā)現(xiàn)event_time字段竟然變成了凌晨3點(diǎn)。問題出在清洗任務(wù)——我用的是unix_timestamp($data.sensor_time) - unix_timestamp($data.receive_time)來判斷時(shí)鐘偏差大于10分鐘就改用接收時(shí)間。但有個(gè)批次的網(wǎng)關(guān)固件有Bug每次重啟后本地時(shí)鐘會(huì)回退8小時(shí)。這些設(shè)備上報(bào)的sensor_time比服務(wù)器的receive_time晚8小時(shí)——注意是晚數(shù)值小絕對(duì)值差剛好480分鐘。我的代碼里判斷條件是絕對(duì)值大于600秒只能識(shí)別“設(shè)備時(shí)間超前”識(shí)別不了“設(shè)備時(shí)間落后8小時(shí)”這種情況。結(jié)果這批設(shè)備的所有數(shù)據(jù)都被當(dāng)成漂移數(shù)據(jù)處理用接收時(shí)間替換了傳感器時(shí)間可接收時(shí)間卻是服務(wù)器收到的時(shí)刻——凌晨3點(diǎn)。修復(fù)思路把“基于單條記錄的時(shí)鐘漂移判斷”改成“基于設(shè)備維度的連續(xù)漂移監(jiān)控”。對(duì)每臺(tái)設(shè)備持續(xù)統(tǒng)計(jì)receive_time - sensor_time的差值分布如果這個(gè)差值在一段時(shí)間內(nèi)穩(wěn)定在一個(gè)非0值附近就說明設(shè)備時(shí)鐘存在固定偏移應(yīng)該按補(bǔ)償量修正而不是直接丟棄。另外最終判斷永遠(yuǎn)以業(yè)務(wù)上“用電高峰在白天”這條規(guī)則校驗(yàn)數(shù)據(jù)是否符合正常模式——這類簡(jiǎn)單的業(yè)務(wù)合理性檢查往往能最快發(fā)現(xiàn)問題。5.2 坑二KafkaFlume寫入HDFS導(dǎo)致的小文件堆成山小文件問題是Hadoop環(huán)境里的經(jīng)典殺手。當(dāng)時(shí)用的Flume從Kafka拉數(shù)據(jù)默認(rèn)每500條提交一次每提交一次就到HDFS寫一個(gè)文件——數(shù)據(jù)量一大一天就產(chǎn)生幾萬個(gè)“小文件”。HDFS的NameNode每個(gè)文件大約占150字節(jié)元數(shù)據(jù)內(nèi)存幾千萬個(gè)文件就能吃掉幾個(gè)GB內(nèi)存更重要的是Spark/Hive跑分析時(shí)要列出和處理幾十萬個(gè)文件光打開文件的時(shí)間就比計(jì)算時(shí)間長。我定位到的根因是兩層Flume的batchSize設(shè)得太小500條hdfsSink的分區(qū)策略又按時(shí)間分了太細(xì)。解決過程用了三板斧加大Flume的batchSize到1000~5000讓每個(gè)批次攢更多數(shù)據(jù)再提交。調(diào)整hdfsSink的rollInterval和rollSize——不要讓文件每幾分鐘就滾動(dòng)一次設(shè)置成“文件超過128MB或30分鐘才滾一次”充分利用大塊寫入保證吞吐。上完這兩步還沒根治后來加了一層HBase做緩沖層數(shù)據(jù)先寫入HBaseHBase再定期合并Compaction后輸出HFile到HDFS完全繞開“Kafka到HDFS直寫”的小文件問題。現(xiàn)在這個(gè)項(xiàng)目中我們最推薦的做法是Kafka → Spark Streaming → HDFS的方式寫入時(shí)直接用coalesce控制輸出分區(qū)數(shù)強(qiáng)制生成足夠大的文件。同一批數(shù)據(jù)不控制分區(qū)數(shù)可能生成幾百個(gè)小文件coalesce(4)直接把輸出收斂為4個(gè)大文件。問題看上去是“文件多”本質(zhì)是“提交粒度太碎”理解了這一點(diǎn)就明白各種方案的本質(zhì)都指向同一個(gè)方向——提高單文件體積。5.3 坑三寫入熱點(diǎn)——RowKey設(shè)計(jì)失敗導(dǎo)致HBase單節(jié)點(diǎn)被打爆HBase處理實(shí)時(shí)數(shù)據(jù)時(shí)RowKey設(shè)計(jì)是性命攸關(guān)的事情。我們?cè)苯佑昧嗽O(shè)備ID 時(shí)間戳作RowKey?？雌饋頉]毛病——但傳感器數(shù)據(jù)的時(shí)間是持續(xù)遞增的于是所有寫入都集中在同一個(gè)Region Server上。那是單Region熱點(diǎn)直接導(dǎo)致某個(gè)節(jié)點(diǎn)負(fù)載極高其他節(jié)點(diǎn)閑著。修復(fù)方案是把RowKey改成設(shè)備ID逆序 時(shí)間戳。比如設(shè)備ID從device_000000000001變成100000000000_device再拼上時(shí)間戳。這樣相同設(shè)備的不同時(shí)間點(diǎn)在主鍵上分布到了不同的Region寫壓力同時(shí)分?jǐn)偟郊豪锏亩鄠€(gè)節(jié)點(diǎn)上。另一個(gè)常見備選方案是在RowKey前面加一個(gè)隨機(jī)前綴比如把device_id哈希后取前兩位但這么做的代價(jià)是查詢時(shí)不知道前綴Scan就不連貫。設(shè)備ID逆序是“分散寫”和“方便查”之間比較平衡的方案——反查時(shí)我們知道完整的設(shè)備ID照樣能直接命中RowKey前綴不會(huì)犧牲查詢性能。修復(fù)之后監(jiān)控圖上寫吞吐從“一個(gè)Region扛”變成“幾個(gè)Region平攤”P99延遲從200ms降到40ms。這個(gè)經(jīng)驗(yàn)后來也驗(yàn)證了設(shè)計(jì)中另一個(gè)判斷物聯(lián)網(wǎng)高吞吐場(chǎng)景下熱點(diǎn)問題本來就是第一殺手RowKey設(shè)計(jì)永遠(yuǎn)要先想寫熱點(diǎn)再想查便利性。6. 完整案例復(fù)盤一個(gè)百萬級(jí)環(huán)境監(jiān)測(cè)平臺(tái)從數(shù)據(jù)接入到分析落地的全過程前面講了這么多方法論和坑最后用一個(gè)我參與過的真實(shí)項(xiàng)目把這些串起來。這個(gè)項(xiàng)目不需要透露具體甲方名字就講場(chǎng)景和數(shù)據(jù)。項(xiàng)目背景幾十個(gè)城市、上萬個(gè)監(jiān)測(cè)點(diǎn)每個(gè)監(jiān)測(cè)點(diǎn)部署PM2.5、溫濕度、風(fēng)速風(fēng)向、噪聲等7種傳感器每10秒上報(bào)一條數(shù)據(jù)。全部設(shè)備累計(jì)每天產(chǎn)生約8億條數(shù)據(jù)單條原始報(bào)文是一條JSON字符串大小約300字節(jié)。這么算下來一天原始數(shù)據(jù)約24GB一年接近9TB。目標(biāo)有三實(shí)時(shí)大屏分鐘級(jí)展示各城市平均PM2.5實(shí)時(shí)告警濃度超閾值觸發(fā)預(yù)警離線分析按周/月輸出空氣質(zhì)量趨勢(shì)報(bào)告評(píng)估不同區(qū)域污染源影響歷史追溯對(duì)任意監(jiān)測(cè)點(diǎn)查詢?nèi)我膺^去一天的分鐘級(jí)曲線響應(yīng)5秒內(nèi)。6.1 數(shù)據(jù)流轉(zhuǎn)鏈路和核心參數(shù)采集端網(wǎng)關(guān)統(tǒng)一上報(bào)到MQTT BrokerBroker直接轉(zhuǎn)發(fā)到Kafka的sensor_raw主題。不直接讓設(shè)備連Kafka因?yàn)镵afka是TCP協(xié)議設(shè)備用MQTT連接生態(tài)更成熟。這就是協(xié)議轉(zhuǎn)換標(biāo)準(zhǔn)做法設(shè)備→MQTT平臺(tái)內(nèi)部→Kafka兩者在Broker層對(duì)接。接入清洗Spark Structured Streaming消費(fèi)sensor_raw做去重、值域校驗(yàn)、時(shí)鐘漂移修正并補(bǔ)上event_date分區(qū)字段用event_time轉(zhuǎn)換寫入HDFS的ODS層按天分區(qū)。實(shí)時(shí)部分同一個(gè)流任務(wù)同時(shí)做分鐘級(jí)聚合寫入HBase Redis緩存最近5分鐘數(shù)據(jù)供大屏點(diǎn)查。離線分析夜里用Spark SQL從ODS層讀全量原始數(shù)據(jù)做多維度聚合到DWD層按城市、小時(shí)、測(cè)點(diǎn)類型分層報(bào)表和趨勢(shì)分析直接查DWD層秒級(jí)響應(yīng)。Kafka的配置也有講究——關(guān)鍵topic分區(qū)數(shù)設(shè)為24個(gè)等于Broker數(shù)保證寫入不傾斜且消費(fèi)并發(fā)可以到24。HDFS塊大小用默認(rèn)的128MB清洗任務(wù)輸出文件控制在128MB以上這個(gè)目標(biāo)所以Spark寫入時(shí)用repartition(24)。6.2 上線的時(shí)踩過的三個(gè)小問題第一個(gè)問題是MQTT到Kafka的鏈路丟數(shù)據(jù)。設(shè)備重連時(shí)網(wǎng)關(guān)會(huì)補(bǔ)傳斷點(diǎn)數(shù)據(jù)但Broker轉(zhuǎn)發(fā)到Kafka時(shí)又用了異步send缺少ACK確認(rèn)壓力大時(shí)有幾條消息被丟掉。排查后發(fā)現(xiàn)是Broker的QoS設(shè)置是“發(fā)完不管”改成QoS1至少一次和Kafka的ackall補(bǔ)上了丟棄缺口。第二個(gè)是Spark處理數(shù)據(jù)傾斜。某幾個(gè)工業(yè)區(qū)的監(jiān)測(cè)點(diǎn)數(shù)量明顯多于普通區(qū)域按區(qū)域聚合時(shí)出現(xiàn)了少數(shù)組件處理多倍數(shù)據(jù)的現(xiàn)象。解決辦法是按“城市小時(shí)”做二次worker分配——代價(jià)是多一次shuffle但從結(jié)果看這點(diǎn)額外消耗完全值得。第三個(gè)是HBase的Compaction風(fēng)暴。高峰期大量Region同時(shí)做Compaction導(dǎo)致某個(gè)RegionServer的IO被打滿反而降低查詢TPS。后來錯(cuò)峰合并把HBase的自動(dòng)Compaction關(guān)掉每天凌晨用運(yùn)維腳本按RegionServer逐個(gè)手動(dòng)執(zhí)行Major Compaction問題就平了。這也是運(yùn)維層面一個(gè)合理的取舍——犧牲一點(diǎn)自動(dòng)性換取全天服務(wù)穩(wěn)定。6.3 大盤數(shù)據(jù)長什么樣上線穩(wěn)定幾周后我摘過幾個(gè)關(guān)鍵數(shù)據(jù)全鏈路端到端延遲平均30秒左右傳感器到平臺(tái)可查大屏數(shù)據(jù)秒級(jí)更新離線每天刷數(shù)任務(wù)2小時(shí)完成8億條原始數(shù)據(jù)Hive查詢分鐘級(jí)的趨勢(shì)分析0.8秒到3秒存儲(chǔ)空間方面ODS層原始數(shù)據(jù)壓縮后4.3GB/天DWD層聚合數(shù)據(jù)僅500MB/天但已經(jīng)能滿足95%以上的查詢需求。這個(gè)壓比其實(shí)很典型——原始數(shù)據(jù)全量保存一份便宜精細(xì)分析集做一份快兩層都保住。7. 一些經(jīng)驗(yàn)沉淀給后續(xù)做同類項(xiàng)目的人幾點(diǎn)忠告全項(xiàng)目走完一遍我個(gè)人最大的體會(huì)有三條第一“接入容易治理難”。傳感器數(shù)據(jù)的接入環(huán)節(jié)看上去每個(gè)設(shè)備都連好就行實(shí)際上治理工作要占整個(gè)項(xiàng)目的70%以上——而且這些工作如果沒有事先規(guī)劃等到數(shù)據(jù)上線之后再做成本和被動(dòng)程度都遠(yuǎn)超想象。清洗規(guī)則、主鍵設(shè)計(jì)、時(shí)間口徑應(yīng)該在數(shù)據(jù)流向設(shè)計(jì)階段就定下來而不是上線了再返工。第二“能落到HDFS的絕不浪費(fèi)在內(nèi)存”。物聯(lián)網(wǎng)數(shù)據(jù)的體量決定了內(nèi)存很貴把幾百億行數(shù)據(jù)長期放Redis或內(nèi)存數(shù)據(jù)庫財(cái)力上往往撐不住。HDFS和對(duì)象存儲(chǔ)都用壓縮配合是最劃算的歷史存儲(chǔ)方式。而內(nèi)存、SSD這些高速資源只留給最需要點(diǎn)查和分析的層。第三“所有問題都能在監(jiān)控里現(xiàn)原形”。我們吃了不少“數(shù)據(jù)錯(cuò)了但沒人發(fā)現(xiàn)”的虧——不是任務(wù)報(bào)錯(cuò)而是結(jié)果不合常識(shí)。后來給Kafka lag、HDFS文件數(shù)、Region熱點(diǎn)、Spark任務(wù)失敗率都做了實(shí)時(shí)監(jiān)控和告警并對(duì)“峰值溫度是否異常偏移”這類業(yè)務(wù)結(jié)果設(shè)置了規(guī)則校驗(yàn)。數(shù)據(jù)平臺(tái)最怕的不是壞是壞了沒人知道。這個(gè)架構(gòu)可能不是最優(yōu)解但它是經(jīng)過了生產(chǎn)環(huán)境驗(yàn)證和故障打磨的。物聯(lián)網(wǎng)數(shù)據(jù)處理沒有銀彈核心是抓住自己的場(chǎng)景特性高吞吐、時(shí)間序列、多源異構(gòu)、低價(jià)值密度——把存儲(chǔ)、清洗、分析都圍繞這四個(gè)特征去設(shè)計(jì)整體不會(huì)跑偏。