16-时序数据库存储方案农业设备时序数据落地、趋势曲线展示写在前面前几篇我们走通了从Modbus采集到MQTT上报的完整链路数据已经到了云端。但数据往哪儿存如果你第一反应是MySQL那就得注意了——农业传感器每10秒采一次10个点位就是每天86400条记录。一个月下来260万条MySQL查个趋势曲线能卡到你怀疑人生。这时候就需要时序数据库了。本篇以InfluxDB为例讲清楚农业时序数据怎么存、怎么查、怎么展示。一、农业时序数据的特点在选数据库之前先搞清楚数据特征特征说明对存储的要求高频写入10秒/次多点位并发写入写入性能要高不能锁表多点位温度、湿度、土壤、CO2、EC等几十个点位标签维度管理方便过滤时间戳依赖每条数据都必须带时间戳时间索引高效支持时间范围查询旧数据降采样近期看秒级历史看小时级/天级支持自动降采样和数据过期读多写少但读模式固定主要是按时间范围聚合查询聚合函数高效均值/最大/最小/滑动窗口数据量线性增长不删数据持续写入压缩率高存储成本低MySQL是为事务设计的时序数据库是为写多读少、按时间聚合设计的。用MySQL存时序数据就像用卡车跑F1——能跑但不是那块料。二、时序数据库选型2.1 主流时序数据库对比特性InfluxDBTDengineTimescaleDBOpenTSDB底层引擎自研LSM自研PostgreSQL扩展HBase部署复杂度低单机开箱即用低中依赖PG高依赖HBase写入性能优秀极佳良好良好SQL支持InfluxQL/Flux类SQL完整SQL有限集群版企业版收费开源支持开源支持开源生态Grafana原生支持Grafana支持Grafana支持Grafana支持中文文档一般优秀国产一般少2.2 选型建议中小项目快速落地选 InfluxDB生态最成熟Grafana无缝对接国产化要求/大规模部署选 TDengine开源集群版中文文档齐全已有PostgreSQL团队选 TimescaleDBSQL语法完全兼容PG超大规模/已有HBase选 OpenTSDB本篇选InfluxDB 2.x它的Flux查询语言和Grafana配合最丝滑。三、InfluxDB核心概念3.1 关键术语InfluxDB概念类比MySQL说明BucketDatabase数据库存储数据的容器MeasurementTable表一组时序数据的集合Tag索引列带索引的维度字段用于过滤如设备IDField普通列不带索引的数值字段存储实际数据如温度值Timestamp主键时间戳每条数据的必备字段Retention Policy无直接对应数据保留策略自动过期删除3.2 数据结构示例一条农业传感器数据的InfluxDB行格式Line Protocolsensor_data,device_idgreenhouse_01,pointtemperature value25.3 1692345600000000000 ↑ ↑ ↑ ↑ measurement Tag(索引) Field(值) Timestamp(ns)3.3 Tag vs Field 的选择这是个容易踩坑的点字段类型特点适用场景Tag字符串类型有索引查询快设备ID、点位名称、大棚编号Field支持多种类型无索引不能做过滤条件温度值、湿度值、电压核心原则你要用来做WHERE过滤条件的就设成Tag只用来展示的数值就设成Field。把温度值设成Tag是没意义的——你不会查value25.3的数据。四、数据模型设计4.1 农业设备点位表映射将网关上报的JSON数据映射到InfluxDB网关上报的JSON: { device_id: greenhouse_01, timestamp: 2026-08-18T10:30:0008:00, data: { temperature: 25.3, humidity: 65.2, soil_moisture: 42.0, co2: 580, ec: 1200 } } 映射到InfluxDB5条记录: sensor_data,device_idgreenhouse_01,pointtemperature value25.3 sensor_data,device_idgreenhouse_01,pointhumidity value65.2 sensor_data,device_idgreenhouse_01,pointsoil_moisture value42.0 sensor_data,device_idgreenhouse_01,pointco2 value580 sensor_data,device_idgreenhouse_01,pointec value12004.2 为什么用一行一个点位而不是一行所有点位方案优点缺点一行一个点位查询灵活新增点位无需改表结构数据行数多一行所有点位行数少新增点位要改写入逻辑空值浪费空间农业场景点位会频繁增减今天加个光照传感器明天加个pH传感器用一行一个点位方案扩展性最好。五、数据写入5.1 Python写入示例frominfluxdb_clientimportInfluxDBClient,Point,WritePrecisionfrominfluxdb_client.client.write_apiimportSYNCHRONOUSimportjsonfromdatetimeimportdatetime,timezone,timedelta TZtimezone(timedelta(hours8))# InfluxDB连接配置INFLUX_CONFIG{url:http://localhost:8086,token:your-api-token-here,org:agri,bucket:sensor_data,}classInfluxWriter:def__init__(self,config):self.clientInfluxDBClient(urlconfig[url],tokenconfig[token],orgconfig[org])self.write_apiself.client.write_api(write_optionsSYNCHRONOUS)self.bucketconfig[bucket]defwrite_sensor_data(self,device_id,data_dict,timestampNone):写入传感器数据iftimestampisNone:timestampdatetime.now(TZ)points[]forpoint_name,valueindata_dict.items():pPoint(sensor_data)\.tag(device_id,device_id)\.tag(point,point_name)\.field(value,float(value))\.time(timestamp,WritePrecision.S)points.append(p)self.write_api.write(bucketself.bucket,orgINFLUX_CONFIG[org],recordpoints)print(f[InfluxDB] 写入{len(points)}个点位, device{device_id})defwrite_from_mqtt_payload(self,payload_json):从MQTT收到的JSON直接写入payloadjson.loads(payload_json)self.write_sensor_data(device_idpayload[device_id],data_dictpayload[data],timestampdatetime.fromisoformat(payload[timestamp]))defclose(self):self.client.close()5.2 批量写入优化InfluxDB推荐批量写入而不是一条一条写frominfluxdb_client.client.write_apiimportWriteOptions# 异步批量写入推荐生产环境使用batch_write_apiclient.write_api(write_optionsWriteOptions(batch_size500,# 每500条一批flush_interval10_000,# 或10秒自动flushjitter_interval2_000,# 随机抖动避免并发冲击))批量写入的吞吐量是单条写入的10~50倍。网关端可以先攒一批再发云端收到后批量写入InfluxDB。六、查询示例InfluxDB 2.x使用Flux查询语言。以下是一些常用查询6.1 查询最近1小时温度数据from(bucket: sensor_data) | range(start: -1h) | filter(fn: (r) r._measurement sensor_data) | filter(fn: (r) r.device_id greenhouse_01) | filter(fn: (r) r.point temperature)6.2 查询最近24小时每小时均温降采样from(bucket: sensor_data) | range(start: -24h) | filter(fn: (r) r._measurement sensor_data) | filter(fn: (r) r.point temperature) | aggregateWindow(every: 1h, fn: mean)aggregateWindow是时序数据库的杀手锏——把高频数据按时间窗口聚合1秒一条变1小时一条查询速度直接快几个数量级。6.3 滑动窗口平均值平滑曲线from(bucket: sensor_data) | range(start: -1h) | filter(fn: (r) r.point temperature) | timedMovingAverage(every: 5m, period: 15m)这条查询的含义每5分钟输出一个值这个值是过去15分钟的平均。效果就是曲线被磨平了适合看趋势。6.4 Python查询示例classInfluxReader:def__init__(self,config):self.clientInfluxDBClient(urlconfig[url],tokenconfig[token],orgconfig[org])self.query_apiself.client.query_api()defquery_temperature(self,device_id,hours1):查询最近N小时的温度数据fluxf from(bucket: sensor_data) | range(start: -{hours}h) | filter(fn: (r) r._measurement sensor_data) | filter(fn: (r) r.device_id {device_id}) | filter(fn: (r) r.point temperature) tablesself.query_api.query(flux)results[]fortableintables:forrecordintable.records:results.append({time:record.get_time().isoformat(),value:record.get_value()})returnresultsdefquery_hourly_avg(self,device_id,point_name,hours24):查询N小时内每小时的均值fluxf from(bucket: sensor_data) | range(start: -{hours}h) | filter(fn: (r) r._measurement sensor_data) | filter(fn: (r) r.device_id {device_id}) | filter(fn: (r) r.point {point_name}) | aggregateWindow(every: 1h, fn: mean) tablesself.query_api.query(flux)results[]fortableintables:forrecordintable.records:results.append({time:record.get_time().isoformat(),avg:round(record.get_value(),2)})returnresults七、Grafana趋势曲线展示7.1 InfluxDB数据源配置打开Grafana → Configuration → Data Sources → Add data source选择 InfluxDB填写URL:http://localhost:8086Organization:agriToken: 你的API TokenDefault Bucket:sensor_dataQuery Language: Flux7.2 温湿度历史曲线面板创建一个Dashboard添加Time Series面板Flux查询温度曲线from(bucket: sensor_data) | range(start: v.timeRangeStart, stop: v.timeRangeStop) | filter(fn: (r) r._measurement sensor_data) | filter(fn: (r) r.device_id greenhouse_01) | filter(fn: (r) r.point temperature) | aggregateWindow(every: v.windowPeriod, fn: mean)面板设置Title:大棚1号 - 温湿度趋势Left Y轴: 温度 (℃)Right Y轴: 湿度 (%)Legend: 显示在底部两条查询一条temperature一条humidity分别映射到左右Y轴7.3 阈值告警线在曲线图上叠加告警阈值线温度超过35℃标红// 温度数据 from(bucket: sensor_data) | range(start: v.timeRangeStart, stop: v.timeRangeStop) | filter(fn: (r) r._measurement sensor_data) | filter(fn: (r) r.point temperature) | aggregateWindow(every: v.windowPeriod, fn: mean) // 阈值线使用Grafana Thresholds功能更简单 // 在面板设置 → Thresholds → 添加 35 为红色阈值或者直接在Grafana面板的Thresholds设置中添加35℃ → 红色高温告警10℃ → 蓝色低温告警Grafana会自动在曲线上画出阈值线数据超过时图表背景变色。7.4 多设备对比面板from(bucket: sensor_data) | range(start: -6h) | filter(fn: (r) r._measurement sensor_data) | filter(fn: (r) r.point temperature) | filter(fn: (r) r.device_id ~ /greenhouse_0[1-3]/) | aggregateWindow(every: 5m, fn: mean) | group(columns: [device_id])这条查询会同时返回3个大棚的温度曲线Grafana自动用不同颜色区分。7.5 常用Dashboard布局┌─────────────────────────────────────────────────┐ │ 智慧农业监控大盘 │ ├─────────────────┬─────────────┬─────────────────┤ │ 当前温度(仪表盘) │ 当前湿度(仪表盘)│ CO2浓度(仪表盘) │ ├─────────────────┴─────────────┴─────────────────┤ │ 温湿度24小时趋势曲线双Y轴 │ ├─────────────────────────┬───────────────────────┤ │ 土壤湿度7天趋势 │ EC值7天趋势 │ ├─────────────────────────┴───────────────────────┤ │ 设备在线状态表格 │ └─────────────────────────────────────────────────┘八、与MySQL对比的存储方案设计8.1 存储架构对比维度MySQL方案InfluxDB方案写入吞吐~5000条/秒单机~10万条/秒单机存储压缩行存储压缩率低列存储时间压缩压缩率高10~50倍时间范围查询需要B树索引数据量大时慢时间索引优化毫秒级响应聚合查询GROUP BY 聚合函数全表扫描内置降采样只读预聚合数据数据过期需要定时任务DELETE自动Retention Policy过期事务支持ACID事务无事务适合场景业务数据订单/用户/配置时序数据传感器/日志/指标8.2 混合存储方案实际项目中MySQL和InfluxDB通常配合使用┌──────────────────────────────────────┐ │ 应用层 │ ├──────────────┬───────────────────────┤ │ MySQL │ InfluxDB │ │ │ │ │ 设备配置表 │ 传感器时序数据 │ │ 用户/权限 │ 历史趋势记录 │ │ 告警规则 │ 设备运行指标 │ │ 告警记录 │ │ │ 系统日志 │ │ └──────────────┴───────────────────────┘MySQL存设备信息、点位配置、用户权限、告警规则、告警记录InfluxDB存温湿度/土壤/CO2等传感器时序数据8.3 数据生命周期管理InfluxDB的Retention Policy自动管理数据生命周期# 通过InfluxDB API设置bucket的保留时间# 例如原始数据保留30天降采样数据保留1年# 方案# 1. raw_data bucket: 保留30天存10秒粒度的原始数据# 2. downsampled bucket: 保留365天存1小时粒度的聚合数据# 用Flux Task自动降采样每小时执行一次task_flux option task { name: downsample_temperature, every: 1h, } from(bucket: raw_data) | range(start: -task.every) | filter(fn: (r) r._measurement sensor_data) | aggregateWindow(every: 1h, fn: mean) | to(bucket: downsampled, org: agri) 这是时序数据库的经典设计模式热数据存高频原始值冷数据存降采样聚合值。既保证了近期的精细化查询又控制了长期存储成本。总结本篇完整覆盖了农业时序数据从存储到展示的全链路数据特征高频写入、多点位、时间戳依赖、需要降采样选型对比InfluxDB生态成熟、TDengine国产高性能、TimescaleDB兼容SQL数据模型Measurement Tagdevice_id/point Fieldvalue Timestamp写入优化批量写入比单条快10~50倍异步批量写入是生产标配查询技巧aggregateWindow降采样、timedMovingAverage平滑曲线Grafana展示温湿度双Y轴曲线、阈值告警线、多设备对比混合存储MySQL管业务数据InfluxDB管时序数据各司其职生命周期热数据高频存30天冷数据降采样存1年自动过期到这里从Modbus现场通讯到云端时序数据展示的完整技术链路就闭环了。整个系列从Modbus协议基础到异常容错再到后端集成、边缘网关、时序存储构成了一套可以直接落地的智慧农业设备数据采集方案。