企业官网建设流程全解析

16-时序数据库存储方案：农业设备时序数据落地、趋势曲线展示

写在前面

前几篇我们走通了从Modbus采集到MQTT上报的完整链路，数据已经到了云端。但数据往哪儿存？如果你第一反应是MySQL，那就得注意了——农业传感器每10秒采一次，10个点位就是每天86400条记录。一个月下来260万条，MySQL查个趋势曲线能卡到你怀疑人生。

这时候就需要时序数据库了。本篇以InfluxDB为例，讲清楚农业时序数据怎么存、怎么查、怎么展示。

一、农业时序数据的特点

在选数据库之前，先搞清楚数据特征：

MySQL是为事务设计的，时序数据库是为"写多读少、按时间聚合"设计的。用MySQL存时序数据，就像用卡车跑F1——能跑，但不是那块料。

二、时序数据库选型

2.1 主流时序数据库对比

2.2 选型建议

• 中小项目快速落地：选 InfluxDB，生态最成熟，Grafana无缝对接
• 国产化要求/大规模部署：选 TDengine，开源集群版，中文文档齐全
• 已有PostgreSQL团队：选 TimescaleDB，SQL语法完全兼容PG
• 超大规模/已有HBase：选 OpenTSDB

本篇选InfluxDB 2.x，它的Flux查询语言和Grafana配合最丝滑。

三、InfluxDB核心概念

3.1 关键术语

3.2 数据结构示例

一条农业传感器数据的InfluxDB行格式（Line Protocol）：

sensor_data,device_id=greenhouse_01,point=temperature value=25.3 1692345600000000000 ↑ ↑ ↑ ↑ measurement Tag(索引) Field(值) Timestamp(ns)

3.3 Tag vs Field 的选择

这是个容易踩坑的点：

核心原则：你要用来做WHERE过滤条件的就设成Tag，只用来展示的数值就设成Field。把温度值设成Tag是没意义的——你不会查"value=25.3"的数据。

四、数据模型设计

4.1 农业设备点位表映射

将网关上报的JSON数据映射到InfluxDB：

网关上报的JSON: { "device_id": "greenhouse_01", "timestamp": "2026-08-18T10:30:00+08:00", "data": { "temperature": 25.3, "humidity": 65.2, "soil_moisture": 42.0, "co2": 580, "ec": 1200 } } 映射到InfluxDB（5条记录）: sensor_data,device_id=greenhouse_01,point=temperature value=25.3 sensor_data,device_id=greenhouse_01,point=humidity value=65.2 sensor_data,device_id=greenhouse_01,point=soil_moisture value=42.0 sensor_data,device_id=greenhouse_01,point=co2 value=580 sensor_data,device_id=greenhouse_01,point=ec value=1200

4.2 为什么用"一行一个点位"而不是"一行所有点位"

农业场景点位会频繁增减（今天加个光照传感器，明天加个pH传感器），用"一行一个点位"方案扩展性最好。

五、数据写入

5.1 Python写入示例

frominfluxdb_clientimportInfluxDBClient,Point,WritePrecisionfrominfluxdb_client.client.write_apiimportSYNCHRONOUSimportjsonfromdatetimeimportdatetime,timezone,timedelta TZ=timezone(timedelta(hours=8))# InfluxDB连接配置INFLUX_CONFIG={"url":"http://localhost:8086","token":"your-api-token-here","org":"agri","bucket":"sensor_data",}classInfluxWriter:def__init__(self,config):self.client=InfluxDBClient(url=config["url"],token=config["token"],org=config["org"])self.write_api=self.client.write_api(write_options=SYNCHRONOUS)self.bucket=config["bucket"]defwrite_sensor_data(self,device_id,data_dict,timestamp=None):"""写入传感器数据"""iftimestampisNone:timestamp=datetime.now(TZ)points=[]forpoint_name,valueindata_dict.items():p=Point("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(bucket=self.bucket,org=INFLUX_CONFIG["org"],record=points)print(f"[InfluxDB] 写入{len(points)}个点位, device={device_id}")defwrite_from_mqtt_payload(self,payload_json):"""从MQTT收到的JSON直接写入"""payload=json.loads(payload_json)self.write_sensor_data(device_id=payload["device_id"],data_dict=payload["data"],timestamp=datetime.fromisoformat(payload["timestamp"]))defclose(self):self.client.close()

5.2 批量写入优化

InfluxDB推荐批量写入，而不是一条一条写：

frominfluxdb_client.client.write_apiimportWriteOptions# 异步批量写入（推荐生产环境使用）batch_write_api=client.write_api(write_options=WriteOptions(batch_size=500,# 每500条一批flush_interval=10_000,# 或10秒自动flushjitter_interval=2_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.client=InfluxDBClient(url=config["url"],token=config["token"],org=config["org"])self.query_api=self.client.query_api()defquery_temperature(self,device_id,hours=1):"""查询最近N小时的温度数据"""flux=f''' 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") '''tables=self.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,hours=24):"""查询N小时内每小时的均值"""flux=f''' 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) '''tables=self.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数据源配置

1. 打开Grafana → Configuration → Data Sources → Add data source
2. 选择 InfluxDB
3. 填写：
   • URL:http://localhost:8086
   • Organization:agri
   • Token: 你的API Token
   • Default Bucket:sensor_data
   • Query Language: Flux

7.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 存储架构对比

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") '''

这是时序数据库的经典设计模式：热数据存高频原始值，冷数据存降采样聚合值。既保证了近期的精细化查询，又控制了长期存储成本。

总结

本篇完整覆盖了农业时序数据从存储到展示的全链路：

1. 数据特征：高频写入、多点位、时间戳依赖、需要降采样
2. 选型对比：InfluxDB生态成熟、TDengine国产高性能、TimescaleDB兼容SQL
3. 数据模型：Measurement + Tag（device_id/point）+ Field（value）+ Timestamp
4. 写入优化：批量写入比单条快10~50倍，异步批量写入是生产标配
5. 查询技巧：aggregateWindow降采样、timedMovingAverage平滑曲线
6. Grafana展示：温湿度双Y轴曲线、阈值告警线、多设备对比
7. 混合存储：MySQL管业务数据，InfluxDB管时序数据，各司其职
8. 生命周期：热数据高频存30天，冷数据降采样存1年，自动过期

到这里，从Modbus现场通讯到云端时序数据展示的完整技术链路就闭环了。整个系列从Modbus协议基础到异常容错，再到后端集成、边缘网关、时序存储，构成了一套可以直接落地的智慧农业设备数据采集方案。