智慧農業(yè)數(shù)據(jù)業(yè)務落地溫濕度/土壤EC/PH值時序數(shù)據(jù)入庫作者黒漂技術佬前面的文章把消息通路搭好了——MQTT 接到設備數(shù)據(jù)SpringBoot 接入微服務分發(fā)。但數(shù)據(jù)最終要去哪總不能光在消息隊列里兜圈子吧。這篇文章聚焦「數(shù)據(jù)落地」——傳感器數(shù)據(jù)怎么入庫、存什么數(shù)據(jù)庫、怎么查詢、怎么管理數(shù)據(jù)生命周期。這是智慧農業(yè)平臺真正產(chǎn)生業(yè)務價值的一步。一、智慧農業(yè)數(shù)據(jù)全景先看看一個大棚里到底有多少種數(shù)據(jù)在流動1.1 環(huán)境數(shù)據(jù)高頻采集通常5~30秒一次傳感器類型采集參數(shù)典型范圍單位空氣溫濕度溫度、濕度-20~60℃ / 0~100%℃ / %RH光照傳感器光照強度0~200000LuxCO2傳感器CO2濃度0~5000ppm風速風向風速、風向0~30m/s1.2 土壤數(shù)據(jù)中頻采集通常1~5分鐘一次傳感器類型采集參數(shù)說明土壤溫濕度土壤溫度、體積含水量埋在不同深度 (10cm/30cm/50cm)土壤EC值電導率反映土壤鹽分濃度0~20 mS/cm土壤PH值酸堿度0~14多數(shù)作物適宜 5.5~7.5氮磷鉀NPK養(yǎng)分含量速效氮/磷/鉀mg/kgEC值Electrical Conductivity電導率是很多新手會忽略的重要指標。它反映的是土壤溶液中可溶性鹽的濃度。EC值過高說明土壤鹽分超標植物根系會「燒」掉——就像你把植物種在鹽水里一樣。水肥一體機施肥時尤其要監(jiān)控 EC 值不然肥施多了比不施還糟糕。1.3 設備狀態(tài)數(shù)據(jù)水泵開關狀態(tài)、水肥機流量、卷簾機位置開了百分之多少、補光燈功率等。這些數(shù)據(jù)和傳感器數(shù)據(jù)不同——它們通常是事件驅動的狀態(tài)變了才上報而非固定頻率采集。二、數(shù)據(jù)庫技術選型不同數(shù)據(jù)放不同數(shù)據(jù)庫這是數(shù)據(jù)庫選型的第一原則。別指望一種數(shù)據(jù)庫通吃所有場景——那和用一把螺絲刀修所有電器沒什么區(qū)別。數(shù)據(jù)類型推薦數(shù)據(jù)庫原因時序數(shù)據(jù)溫濕度、EC等InfluxDB / TDengine專為時序優(yōu)化寫入快、壓縮高、聚合查詢強關系數(shù)據(jù)設備臺賬、告警記錄MySQL / PostgreSQL事務支持、復雜關聯(lián)查詢日志數(shù)據(jù)MongoDB / ElasticsearchSchema靈活、全文檢索重點說說時序數(shù)據(jù)庫。普通關系型數(shù)據(jù)庫存時序數(shù)據(jù)有兩個致命缺陷1. 寫入性能不足。1000 個傳感器每秒各寫一條數(shù)據(jù)一天就是 8640 萬條。MySQL 的 BTree 索引在這種寫入壓力下會頻繁分裂性能急劇下降。2. 存儲浪費巨大。時序數(shù)據(jù)高度重復設備 ID、傳感器類型等冗余嚴重。時序數(shù)據(jù)庫用列式存儲 差值編碼能把數(shù)據(jù)壓縮到原始大小的十分之一。選 InfluxDB 還是 TDengine簡單說如果你的團隊熟悉 SQLTDengine 更友好標準 SQL 超級表模型如果已經(jīng)有 InfluxDB 運維經(jīng)驗或者對 Flux 查詢語言不排斥InfluxDB 生態(tài)更成熟。本文以 InfluxDB 為例。三、MQTT消息到數(shù)據(jù)入庫的完整鏈路MQTT消息到達 │ ▼ 解析 JSON → 提取字段 │ deviceId, sensorType, value, timestamp ▼ 數(shù)據(jù)清洗異常值剔除 │ 溫度突變 10℃標記異常不入庫 │ 濕度 100%明顯錯誤丟棄 ▼ 構建數(shù)據(jù)模型 │ Point.measurement(sensor_data) │ .tag(device_id, s001) │ .tag(sensor_type, temperature) │ .addField(value, 28.5) ▼ 批量寫入 InfluxDB關鍵代碼實現(xiàn)ServicepublicclassSensorDataPersistenceService{AutowiredprivateInfluxDBinfluxDB;// 批量緩沖區(qū)攢夠100條或者滿5秒就寫入privatefinalListPointbuffernewArrayList();privatelonglastFlushTimeSystem.currentTimeMillis();privatestaticfinalintBATCH_SIZE100;privatestaticfinallongFLUSH_INTERVAL5000;publicsynchronizedvoidwritePoint(Pointpoint){buffer.add(point);if(buffer.size()BATCH_SIZE||System.currentTimeMillis()-lastFlushTimeFLUSH_INTERVAL){flush();}}privatevoidflush(){if(buffer.isEmpty())return;try{influxDB.write(BATCH_SIZE,TimeUnit.SECONDS,buffer);buffer.clear();lastFlushTimeSystem.currentTimeMillis();}catch(Exceptione){log.error(寫入 InfluxDB 失敗,e);// 寫入失敗不丟數(shù)據(jù)保留在buffer中等待下一次flush}}}為什么要批量寫而不是來一條寫一條InfluxDB 寫入操作是有開銷的——每條 HTTP 請求都涉及網(wǎng)絡往返。100 條數(shù)據(jù)一次寫和 100 次各寫一條性能差距可能是 50 倍以上。這叫batch write批量寫入是時序數(shù)據(jù)庫的標準操作方式。四、InfluxDB 數(shù)據(jù)建模InfluxDB 的數(shù)據(jù)模型用measurement tag field timestamp來描述publicPointbuildSensorPoint(SensorDatadata){returnPoint.measurement(sensor_data)// measurement 表名.time(data.getTimestamp(),TimeUnit.MILLISECONDS)// 時間戳.tag(farm_id,data.getFarmId())// tag 索引列.tag(greenhouse_id,data.getGreenhouseId())// tag支持group by.tag(device_id,data.getDeviceId()).tag(sensor_type,data.getSensorType())// temperature/humidity/ec/ph.addField(value,data.getValue())// field 值列不建索引.addField(unit,data.getUnit()).build();}Tag 和 Field 的區(qū)別很重要Tag會被索引用于WHERE和GROUP BY。適合設備 ID、傳感器類型這類基數(shù)不高的字符串。Field不建索引存具體數(shù)值。適合溫度值、濕度值這類持續(xù)變化的數(shù)值。一個常見錯誤是把傳感器數(shù)值也設為 Tag——這會導致 InfluxDB 索引急劇膨脹查詢反而變慢。Tag 是用來「找數(shù)據(jù)」的Field 才是真正的「數(shù)據(jù)」。五、數(shù)據(jù)查詢API設計有了數(shù)據(jù)還得給前端大屏、管理后臺提供查詢接口RestControllerRequestMapping(/api/sensor)publicclassSensorDataController{AutowiredprivateInfluxDBinfluxDB;/** * 查詢某大棚某傳感器的時間段溫度曲線 * GET /api/sensor/query?greenhouseIdgh01sensorTypetemperaturestartxxxendxxx */GetMapping(/query)publicListDataPointquery(RequestParamStringgreenhouseId,RequestParamStringsensorType,RequestParamlongstart,RequestParamlongend){QueryquerynewQuery(String.format(SELECT MEAN(\value\) as \value\ FROM \sensor_data\ WHERE \greenhouse_id\%s AND \sensor_type\%s AND time %dms AND time %dms GROUP BY time(1m) fill(previous),greenhouseId,sensorType,start,end),smart_agriculture// 數(shù)據(jù)庫名);QueryResultresultinfluxDB.query(query);returnconvertToDataPoints(result);}/** * 最近1小時各傳感器均值 */GetMapping(/latest-hour-avg)publicMapString,DoublelatestHourAvg(RequestParamStringgreenhouseId){// 查詢最近的溫度均值和濕度均值QuerytempQuerynewQuery(String.format(SELECT MEAN(\value\) FROM \sensor_data\ WHERE \greenhouse_id\%s AND \sensor_type\temperature AND time now() - 1h,greenhouseId),smart_agriculture);// 類似查詢濕度...// 返回 {temperature: 26.5, humidity: 72.3}returnresult;}}GROUP BY time(1m)是按 1 分鐘聚合這在畫折線圖時非常有用——原始數(shù)據(jù)可能每秒都有但圖表只需要分鐘級別的粒度。fill(previous)是指定聚合窗口沒有數(shù)據(jù)時用上一個窗口的值填充避免折線圖出現(xiàn)斷崖。六、數(shù)據(jù)保留策略全存是不可能全存的每天幾千萬條傳感器數(shù)據(jù)硬盤再大也扛不住。需要制定分層保留策略-- InfluxDB 保留策略配置-- 原始數(shù)據(jù)保留30天CREATERETENTION POLICYraw_30dONsmart_agricultureDURATION30dREPLICATION1DEFAULT;-- 小時聚合數(shù)據(jù)保留1年通過連續(xù)查詢自動生成CREATECONTINUOUS QUERYcq_hourly_avgONsmart_agricultureBEGINSELECTMEAN(value)ASvalueINTOsmart_agriculture.hourly.sensor_data_hourlyFROMsmart_agriculture.raw_30d.sensor_dataGROUPBYtime(1h),*END;-- 日聚合數(shù)據(jù)永久保留CREATERETENTION POLICYforeverONsmart_agricultureDURATION INFREPLICATION1;這就像視頻監(jiān)控——原始錄像存 30 天關鍵片段永久保留。30 天后一秒一次的溫度數(shù)據(jù)被聚合為 1 小時均值、1 日均值。查詢去年今天的溫度看日聚合就夠了沒必要翻秒級原始數(shù)據(jù)。連續(xù)查詢Continuous Query是 InfluxDB 的殺手級特性。它自動按固定頻率比如每小時執(zhí)行聚合 SQL把結果寫入另一個 measurement。你完全不需要寫定時任務代碼?？偨Y智慧農業(yè)數(shù)據(jù)落地的關鍵決策就三個時序數(shù)據(jù)走時序庫——別用 MySQL 存?zhèn)鞲衅鲾?shù)據(jù)那是自討苦吃批量寫入——100 條一寫比 100 次各寫 1 條快一個數(shù)量級分層保留——原始數(shù)據(jù)定時清聚合數(shù)據(jù)長期存數(shù)據(jù)入庫不是終點而是業(yè)務價值的起點——有了數(shù)據(jù)你才能畫曲線、做分析、觸發(fā)告警、訓練模型。