物聯(lián)網(wǎng)數(shù)據(jù)這幾年我經(jīng)手了不少從車聯(lián)網(wǎng)終端上報的軌跡點到廠房里各種PLC傳感器采集的溫度、振動、能耗數(shù)據(jù)說白了就是一個字多。設(shè)備一多、頻率一高一天少說幾千萬條記錄傳統(tǒng)的關(guān)系型數(shù)據(jù)庫根本扛不住——不是存不下是查詢和統(tǒng)計能把人卡到懷疑人生。后來我把整套處理鏈路遷到 Hadoop 生態(tài)上用 HDFS 做統(tǒng)一存儲、Spark 做批量清洗和聚合、Hive 做即席查詢才算把這塊硬骨頭啃了下來。這篇東西不是教科書式科普是我從零搭建、踩坑、上線、調(diào)優(yōu)的一次完整復(fù)盤。不管你是剛接觸大數(shù)據(jù)的學(xué)生還是想給團隊的物聯(lián)網(wǎng)數(shù)據(jù)找個靠譜處理方案都可以照著這里的思路走一遍。1. 物聯(lián)網(wǎng)數(shù)據(jù)與 Hadoop 的適配邏輯1.1 物聯(lián)網(wǎng)數(shù)據(jù)到底“臟”在哪很多人一提到物聯(lián)網(wǎng)第一反應(yīng)是傳感器很智能。但實際拿到手里的物聯(lián)網(wǎng)原始數(shù)據(jù)往往是一堆“半成品”設(shè)備離線產(chǎn)生的補傳數(shù)據(jù)時間戳錯亂信號干擾導(dǎo)致數(shù)值突然跳變不同廠商設(shè)備的字段命名千奇百怪有的上報 JSON、有的只傳 CSV還有的直接把二進制協(xié)議解析后的字符串扔過來。更麻煩的是同一批設(shè)備里還混著大量重復(fù)上報和無效點位。這些數(shù)據(jù)天然有三個特點體量大、亂序多、價值密度低。體量大意味著單機存儲和計算都吃不消亂序多意味著清洗時必須有全局排序和去重邏輯價值密度低意味著我們往往要先把原始明細(xì)存下來之后反復(fù)跑不同口徑的統(tǒng)計不能只留一個壓縮過的匯總表。Hadoop 的 HDFS 恰好能低成本地把原始數(shù)據(jù)全量留存MapReduce 和 Spark 這類批計算引擎又適合做全量掃描和清洗所以從底層架構(gòu)上講物聯(lián)網(wǎng)數(shù)據(jù)放在 Hadoop 生態(tài)里是成立的。1.2 Hadoop 三大組件各管哪一段Hadoop 給人的印象是一個“大箱子”但實際拆開看核心就三塊HDFS、YARN、計算引擎。HDFS 管存儲把大文件切塊分布式地放在多個節(jié)點上并且默認(rèn)復(fù)制三份防止某臺機器掛掉丟數(shù)據(jù)。物聯(lián)網(wǎng)場景下一天的原始日志可以到幾十 GB直接丟 HDFS 里不需要提前設(shè)計什么分庫分表。YARN 管資源它像一個大管家誰要跑計算任務(wù)就給它分配多少 CPU 和內(nèi)存。多個計算引擎可以同時跑在一個集群上互不打架。計算引擎管邏輯老一代是 MapReduce現(xiàn)在生產(chǎn)環(huán)境基本都用 Spark、Hive 或 Flink 來處理。MapReduce 雖然執(zhí)行慢但勝在穩(wěn)定適合跑夜間批量任務(wù)Spark 執(zhí)行效率高適合做多次迭代的清洗和聚合。這套架構(gòu)解決的核心問題是把“一臺機器硬扛”換成“一群機器分工協(xié)作”。你不需要關(guān)心某條數(shù)據(jù)存在哪個節(jié)點的哪塊磁盤上只需要把任務(wù)提交上去框架自己會找數(shù)據(jù)所在的節(jié)點去算也就是常說的 data locality——移動計算而不是移動數(shù)據(jù)。數(shù)據(jù)量越大的時候這個優(yōu)勢越明顯。2. 從零搭建可用的 Hadoop 環(huán)境2.1 單機偽分布式與真實集群的選擇第一次上手 Hadoop沒必要一開始就上五臺物理機先用“偽分布式”把流程跑通是性價比最高的方式。所謂偽分布式就是在一臺 Linux 機器上同時啟動 NameNode、DataNode、ResourceManager、NodeManager 這幾個進程模擬出一個“一節(jié)點集群”。我一般建議新手用 Ubuntu 虛擬機加三到四個節(jié)點來做。偽分布式適合驗證代碼和熟悉命令但如果你要測數(shù)據(jù)分片、節(jié)點宕機后的容錯或者 YARN 的資源調(diào)度最少得搭一個三節(jié)點集群一個 NameNode 節(jié)點和兩個 DataNode 節(jié)點。這一步千萬別圖省事只搭偽分布式因為真實環(huán)境里的很多坑比如數(shù)據(jù)塊副本不足導(dǎo)致文件變成 corrupt 狀態(tài)、節(jié)點之間 hostname 解析不通只有在多節(jié)點下才會暴露。搭建時注意幾個關(guān)鍵點所有節(jié)點使用統(tǒng)一的 Linux 用戶并且配置 SSH 免密登錄。否則每次啟動集群都要輸密碼啟動腳本根本沒法用。/etc/hosts里必須把集群所有節(jié)點的 IP 和 hostname 對應(yīng)關(guān)系寫全。很多人第一次搭集群失敗就是因為主機名解析不對節(jié)點之間互相找不到。配置 Java 環(huán)境變量時Hadoop 3.x 需要 Java 8 以上我建議直接用 Java 8兼容性最穩(wěn)。2.2 核心配置項與內(nèi)存參數(shù)配置文件主要改三個core-site.xml、hdfs-site.xml、yarn-site.xml。新手最容易犯的錯是照搬網(wǎng)上的配置不看自己的內(nèi)存大小導(dǎo)致 NodeManager 申請內(nèi)存超過物理機內(nèi)存啟動后直接被系統(tǒng)殺掉。我個人的經(jīng)驗做法是物理機或虛擬機內(nèi)存 8 GB 的話給 NameNode 分配 1 GBDataNode 1 GBResourceManager 1 GBNodeManager 2 GB剩下留給操作系統(tǒng)和后續(xù)跑的 Spark 任務(wù)。yarn-site.xml里把yarn.nodemanager.resource.memory-mb設(shè)為 2048yarn.nodemanager.resource.cpu-vcores設(shè)為 2這樣至少能穩(wěn)定跑起來兩個容器。偽分布式的核心配置如下!-- core-site.xml -- configuration property namefs.defaultFS/name valuehdfs://localhost:9000/value /property /configuration !-- hdfs-site.xml -- configuration property namedfs.replication/name value1/value /property /configuration !-- yarn-site.xml -- configuration property nameyarn.nodemanager.aux-services/name valuemapreduce_shuffle/value /property property nameyarn.nodemanager.resource.memory-mb/name value2048/value /property /configuration偽分布式里dfs.replication必須設(shè) 1因為只有一臺 DataNode默認(rèn)的 3 份副本會一直處于 under-replicated 狀態(tài)看著很煩。多節(jié)點集群再改回 3。啟動順序別搞錯先start-dfs.sh再start-yarn.sh。然后輸入jps能看到 NameNode、DataNode、ResourceManager、NodeManager 這 4 個進程就說明基礎(chǔ)環(huán)境通了。3. 把物聯(lián)網(wǎng)數(shù)據(jù)送進 HDFS3.1 數(shù)據(jù)落地路線Flume 與腳本導(dǎo)入數(shù)據(jù)從設(shè)備端到 HDFS常見路線有三條設(shè)備端直接推送到 Kafka再由 Flume 或自研消費程序?qū)懭?HDFS。這條鏈路最接近生產(chǎn)環(huán)境適合需要緩沖削峰的場景。設(shè)備端產(chǎn)生的文件按小時落到邊緣網(wǎng)關(guān)網(wǎng)關(guān)上的腳本定時調(diào)用hdfs dfs -put上傳。如果已經(jīng)有采集服務(wù)在寫 MySQL再用 Sqoop 每天把增量數(shù)據(jù)同步到 HDFS。這種做法適合業(yè)務(wù)系統(tǒng)已經(jīng)跑了一段時間、想拿歷史數(shù)據(jù)做分析的團隊。我在一個交通信息分析的小項目里用的是第二種網(wǎng)關(guān)每 5 分鐘生成一個包含車輛 GPS 點位和速度的 JSON 文件然后由一個 shell 定時任務(wù)把文件壓縮成 gzip 后傳到 HDFS 指定目錄。之所以先壓縮再上傳是因為 GPS 軌跡點這類文本數(shù)據(jù)壓縮率極高10 GB 原始數(shù)據(jù) gzip 后往往只剩 1.5 GB 左右能大幅節(jié)省網(wǎng)絡(luò)帶寬和存儲成本。壓縮格式的選擇上我建議優(yōu)先用 gzip。雖然它不支持切分但物聯(lián)網(wǎng)設(shè)備上報處理邏輯簡單每個業(yè)務(wù)目錄下本身就按小時分好了文件單文件多壓縮幾倍后可能只有十幾 MB切分需求不明顯。只有當(dāng)單個文件超過 HDFS 塊大小默認(rèn) 128 MB時你才需要考慮用支持切分的 LZO 或 snappy。3.2 文件格式與壓縮比物聯(lián)網(wǎng)數(shù)據(jù)推薦用列式存儲還是行式存儲我踩過坑這里給個明確建議如果走 Spark SQL 或 Hive最終表用 Parquet snappy 壓縮但如果只是把原始數(shù)據(jù)先原樣留存就保留 JSON 或者 CSV 的 gzip。Parquet 這種列式格式在查詢時只讀需要的列性能比 CSV 高一個量級。但列式存儲的前提是你已經(jīng)完成了數(shù)據(jù)清洗字段統(tǒng)一、類型統(tǒng)一。原始物聯(lián)網(wǎng)數(shù)據(jù)還沒清洗前字段可能有缺失類型也可能不對強行轉(zhuǎn) Parquet 反而會引入很多解析錯誤。所以我一貫的原則是原始層用什么格式無所謂干凈層必須要用 Parquet。把 JSON 轉(zhuǎn)成 Parquet推薦直接用 Spark 讀進來再寫出去幾行代碼就搞定val rawDF spark.read.json(/data/iot/raw/2024/06/01) val cleanedDF rawDF.select( $device_id, $event_time.cast(timestamp), $longitude.cast(double), $latitude.cast(double), $speed.cast(double) ) cleanedDF.write.mode(overwrite) .partitionBy(dt) .format(parquet) .save(/data/iot/clean/2024/06/01)這段代碼的核心邏輯是先加載當(dāng)天的原始 JSON只挑出后續(xù)分析要用到的字段并轉(zhuǎn)換成強類型再按日期分區(qū)寫入 Parquet。一旦走到這一步后面的統(tǒng)計任務(wù)就都在這份干凈數(shù)據(jù)上跑了。4. 用 MapReduce 模型做第一版清洗與聚合4.1 按設(shè)備 ID 聚合的 MapReduce接觸一個新框架第一件事不是背 API而是跑通一個能解決實際問題的例子。網(wǎng)上那些 wordcount 例子用途不大我建議直接寫一個“統(tǒng)計每臺設(shè)備當(dāng)天上報了多少條點位”的作業(yè)流程和 wordcount 一模一樣但更貼近物聯(lián)網(wǎng)場景。任務(wù)很具體從原始 JSON 里提取device_id和event_time按小時統(tǒng)計每臺設(shè)備的點位數(shù)量找出上報異常偏少的設(shè)備。用 MapReduce 實現(xiàn)時Mapper 負(fù)責(zé)解析 JSON把鍵設(shè)為device_id、值設(shè)為 1Reducer 負(fù)責(zé)求和。核心代碼長這樣public class DeviceCountMapper extends MapperObject, Text, Text, IntWritable { private final static IntWritable one new IntWritable(1); private Text deviceId new Text(); public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String line value.toString(); // 每行是一條 JSON解析 field 可以引入 fastjson 或自寫簡單抽取 String device parseDeviceId(line); deviceId.set(device); context.write(deviceId, one); } } public class DeviceCountReducer extends ReducerText, IntWritable, Text, IntWritable { private IntWritable result new IntWritable(); public void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException { int sum 0; for (IntWritable val : values) { sum val.get(); } result.set(sum); context.write(key, result); } }這個作業(yè)跑完后你會得到一個明細(xì)輸出每臺設(shè)備對應(yīng)的上報總數(shù)。但要注意如果某臺設(shè)備一整天都沒上報它根本不會出現(xiàn)在輸出里因為 MapReduce 只處理存在的數(shù)據(jù)。要找出“完全沒上報的設(shè)備”還需要拿設(shè)備元數(shù)據(jù)表做一次外連接這通常放到 Hive 里處理更合適。4.2 任務(wù)提交與日志排查作業(yè)寫好后打包成 jar用hadoop jar命令提交到 YARN 上運行hadoop jar iot-process.jar com.example.DeviceCount \ /data/iot/raw/2024/06/01 \ /data/iot/result/device_count/2024/06/01我每次跑任務(wù)都會先跟蹤一下進度用yarn application -status或直接打開 ResourceManager 的 Web 頁面看 container 日志。新手常見的問題是代碼一跑就內(nèi)存溢出但不知道去哪兒看。MapReduce 里 Mapper 的內(nèi)存溢出通常表現(xiàn)為 task 反復(fù)重試看日志時重點盯著mapreduce.map.memory.mb和mapreduce.reduce.memory.mb這兩個參數(shù)。默認(rèn)值在 3.x 里是 1 GB 左右如果你每條 JSON 里嵌了一個很大的字段比如把設(shè)備上報的整包數(shù)據(jù)都放到 value 里1 GB 很容易被打滿。4.3 合并小文件物聯(lián)網(wǎng)數(shù)據(jù)一落地就容易產(chǎn)生大量小文件網(wǎng)關(guān)每 5 分鐘一個文件一天就是幾百個一個月就是上萬個。小文件多了NameNode 內(nèi)存會崩潰因為每個文件都要在內(nèi)存里維護元數(shù)據(jù)跑任務(wù)時每個文件還要對應(yīng)一個 task啟動 task 的開銷比計算本身還大。我的處理手段是定期用 Spark 做一次“小文件重分區(qū)”spark.read.parquet(/data/iot/clean) .repartition(24) // 按一天 24 小時控制輸出文件數(shù) .write.mode(overwrite) .partitionBy(dt) .option(compression, snappy) .format(parquet) .save(/data/iot/clean_merged)這里repartition(24)的意義是把當(dāng)天數(shù)據(jù)重新打散為固定數(shù)量的 24 個大文件而不是按分區(qū)目錄原樣保留幾百個小文件。但這句話有個前提——你要清楚自己集群的塊大小。如果 24 個文件每個才 5 MB那分得還是太細(xì)。正確的做法是估算當(dāng)天總數(shù)據(jù)量比如 2 GB想讓每個文件在 128 MB 左右那就repartition(16)左右。寧可多試幾次把文件數(shù)調(diào)小也不要每個文件幾 MB那樣跑數(shù)倉任務(wù)會慢得讓人抓狂。5. 升級為 Spark SQL Hive 的工業(yè)級處理5.1 為什么最終遷移到 Spark/HiveMapReduce 作業(yè)寫起來啰嗦而且每次迭代都要把中間結(jié)果寫回磁盤跑多步驟清洗鏈路時慢得惱人。后來我把核心鏈路遷移到了 Spark SQL Hive 上用 Hive 管理表結(jié)構(gòu)用 Spark SQL 做查詢和計算。遷移帶來的收益很明顯Spark 基于內(nèi)存計算多階段的 ETL 任務(wù)比 MapReduce 快 3 到 10 倍。Spark SQL 內(nèi)置大量函數(shù)處理 JSON、時間窗口、開窗去重都比手寫 MapReduce 方便得多。Hive 的元數(shù)據(jù)服務(wù)讓所有表結(jié)構(gòu)統(tǒng)一Spark 可以直接spark.sql(select ... from xxx where ...)不需要自己寫文件路徑解析邏輯。Hive 的另一個隱藏價值是它把存儲路徑變成了類似關(guān)系數(shù)據(jù)庫的表結(jié)構(gòu)。比如數(shù)據(jù)放在 HDFS 的/warehouse/iot.db/device_report目錄下你在 Hive 里建一張外層表指定 location 指向這個目錄執(zhí)行 SQL 時就不用關(guān)心底層文件是怎么組織的了。5.2 建表、分區(qū)與 ZooKeeper 在分布式協(xié)調(diào)中的角色用 Hive 管理物聯(lián)網(wǎng)數(shù)據(jù)的核心是分區(qū)設(shè)計。我常用的分區(qū)字段是dt按天分區(qū)。這樣做的好處是查詢時可以直接裁剪掉無關(guān)分區(qū)只掃需要的日期效率提升非常明顯。建表示例CREATE EXTERNAL TABLE iot.device_report ( device_id STRING, event_time TIMESTAMP, longitude DOUBLE, latitude DOUBLE, speed DOUBLE, temperature DOUBLE ) PARTITIONED BY (dt STRING) STORED AS PARQUET LOCATION /data/iot/warehouse/device_report;這是外表數(shù)據(jù)文件已經(jīng)存在 HDFS 上Hive 只負(fù)責(zé)掛元數(shù)據(jù)。跑任務(wù)前先執(zhí)行一條MSCK REPAIR TABLE device_report;讓 Hive 自動識別 HDFS 上新出現(xiàn)的分區(qū)目錄否則查不到數(shù)據(jù)。這個命令我每次都會忘一次然后被同事吐槽。集群規(guī)模稍微上來以后Hive 和 HBase 這類服務(wù)經(jīng)常同時跑在一個集群上。多節(jié)點分布式環(huán)境下各節(jié)點之間的狀態(tài)同步需要協(xié)調(diào)者。這里就是 ZooKeeper 登場的地方HDFS 的 NameNode 高可用依賴 ZooKeeper 做主備切換YARN 的 ResourceManager 高可用同樣依賴它。實際生產(chǎn)里我見過把 ZooKeeper 單獨用三臺機器搭成一個 ensemble 的也見過和 Hadoop 節(jié)點混部看機器資源情況而定。最關(guān)鍵的配置是zoo.cfg里的server.1、server.2、server.3三行地址要寫對并且每個節(jié)點都要配一個數(shù)據(jù)目錄不能共享一個目錄否則啟動后互相搶鎖狀態(tài)一片混亂。5.3 交通信息統(tǒng)計案例拿我之前做的交通信息分析系統(tǒng)舉例。車輛的 GPS 軌跡數(shù)據(jù)進 HDFS 以后最早是直接查明細(xì)慢得要命。后來改成 Hive 里維護一張“車輛時段聚合表”每天夜間用 Spark 作業(yè)跑一次統(tǒng)計每輛車在每一小時內(nèi)的平均速度、最大速度、軌跡點數(shù)量和經(jīng)過的路段標(biāo)識。核心 SQL 大概是INSERT OVERWRITE TABLE iot.device_hourly_summary PARTITION (dt) SELECT device_id, dt, hour(event_time) AS hour, count(*) AS point_count, round(avg(speed), 2) AS avg_speed, max(speed) AS max_speed FROM iot.device_report WHERE dt 2024-06-01 GROUP BY device_id, dt, hour(event_time);跑完這條 SQL每天的數(shù)據(jù)量從幾千萬條明細(xì)壓縮成幾十萬條匯總后續(xù)做可視化、出報表直接查這張聚合表就行幾十毫秒內(nèi)能出結(jié)果。這個思路是物聯(lián)網(wǎng)數(shù)據(jù)處理里最核心的一招明細(xì)層永遠(yuǎn)保留原始數(shù)據(jù)匯總層只留指標(biāo)。分析需求變的時候從明細(xì)層重新跑一套聚合就行不需要回設(shè)備端重新采集。6. 典型故障排查與常見問題速查6.1 經(jīng)常出現(xiàn)的錯誤跑 Hadoop 和 Spark 的幾個常見問題基本每個人都會碰到Could not find or load main class或者提交作業(yè)后一直顯示RUNNING但沒有任何輸出。多半是 jar 包依賴沒打全或者 worker 節(jié)點上的 Spark 環(huán)境變量沒配一致。我慣用spark-submit --master yarn --deploy-mode cluster時把依賴包打進 fat jar本地模式?jīng)]問題、集群模式就報 ClassNotFound基本都是這個原因。DataNode 啟動后過幾分鐘進程消失。先看日志常見是磁盤空間不足或者dfs.data.dir指向了不存在的目錄。MapReduce 任務(wù) 100% map 完成reduce 一直停在 33.33%。不要慌大概率是 reducer 正在拉取 map 輸出網(wǎng)絡(luò)或磁盤速度慢而已但如果一直卡著不動去查 reducers 節(jié)點是不是內(nèi)存碎片太多。Hive 查詢返回結(jié)果為空明明 HDFS 上有文件。先跑dfs -ls看路徑有沒有權(quán)限問題再MSCK REPAIR刷新分區(qū)最后檢查是不是 iot 表存的是 rename 前的舊路徑。6.2 面試和架構(gòu)設(shè)計里常被問到的點不少讀者是學(xué)生面試大數(shù)據(jù)崗位時經(jīng)常被問“說說你做過的一個 Hadoop 項目”。我建議不要只背理論把一個物聯(lián)網(wǎng)數(shù)據(jù)處理項目講透。面試官常追問的點其實就兩個方向一是你如何處理數(shù)據(jù)傾斜二是你如何保證數(shù)據(jù)不丟不重。數(shù)據(jù)傾斜在物聯(lián)網(wǎng)數(shù)據(jù)分析里非常常見。比如統(tǒng)計設(shè)備活躍度時某幾個頭部設(shè)備的點位特別多reduce 階段就一個 task 卡到天荒地老。解決辦法我之前用的是加鹽打散先把 key 后面拼上一個隨機數(shù)把數(shù)據(jù)分成 10 份分別聚合第二步再按真實 key 聚合一次。代價是跑兩輪任務(wù)但穩(wěn)定性好了很多。數(shù)據(jù)不丟不重里最值得注意的坑是“重跑作業(yè)時覆蓋寫”。寫入 HDFS 時用overwrite模式一定要謹(jǐn)慎如果下游已經(jīng)在讀這份數(shù)據(jù)你覆蓋的同時下游可能讀到半份文件。我的習(xí)慣是每次跑批都先把結(jié)果寫到臨時目錄成功后再把臨時目錄原子地 rename 成正式目錄能避免很多詭異問題。收尾的經(jīng)驗之談要是把整個流程壓縮成一句話那就是物聯(lián)網(wǎng)數(shù)據(jù)處理的關(guān)鍵不在技術(shù)棧有多高級而在于把“原始數(shù)據(jù)沉淀”和“指標(biāo)計算”分層做好。Hadoop 這層架構(gòu)最大的價值是給你提供了一個穩(wěn)定、能擴展的底座讓你不用每天擔(dān)心“機器磁盤是不是又滿了”“數(shù)據(jù)要不要刪一部分騰空間”。我做的項目里從最初的單機腳本到后來三節(jié)點集群加 Hive 加 ZooKeeper 協(xié)調(diào)整個演進過程花了大概兩個月最耗時間的部分反而不是寫代碼而是調(diào)內(nèi)存參數(shù)和排查各種分布式環(huán)境下的怪毛病。最后分享一個小技巧拿到一批物聯(lián)網(wǎng)數(shù)據(jù)先別急著寫清洗邏輯花半小時把數(shù)據(jù)可能存在的異常列個清單——時間亂序、設(shè)備 ID 重復(fù)、空字段、超范圍數(shù)值寫清楚每種異常對應(yīng)什么處理規(guī)則然后才開始建表和寫 ETL。這個清單看起來不起眼但它能讓你少走非常多的彎路也能讓你在跟同事討論需求時更有底氣。