简介这份PPT资源面向农业信息化从业者、智慧园区规划人员及农业项目方案撰写者提供2022年智慧农业示范园区大数据分析平台的整体解决方案。内容围绕大数据分析平台展开涵盖建设现状与需求分析、顶层设计、整体实施方案、应用功能、效益分析及推广方案等模块并细化到领导驾驶舱、大苗情监测、水产与畜牧养殖监测、温室大棚管控、农产品质量安全追溯、远程专家诊断、智慧物流冷链及电商平台等子系统可帮助读者快速理解智慧农业从数据采集到决策支持的完整链路。资源包内含1个pptx文件大小约15.86MB以图文并茂的演示文稿形式呈现便于直接参考或二次编辑。目前已有157人学习下载适合需要搭建智慧农业方案框架、梳理园区功能模块或撰写同类项目材料的技术与管理人员借鉴。1. 智慧农业示范园区大数据分析平台从一份方案书到一个能跑的系统去年帮一个做农业信息化的团队看他们的智慧农业示范园区大数据分析平台他们手里有一份 2022 年做的整体解决方案 PPT页数不少架构图画得也漂亮但真到落地的时候发现一个尴尬事方案里写的“数据采集—清洗—存储—分析—可视化”五层架构每一层都能讲半小时可园区里真正能接进来的数据源只有三样——几个大棚的温湿度传感器、一台老式气象站、还有人工填的 Excel 表。这就是智慧农业示范园区大数据分析平台最常见的现状方案书是完美的数据是残缺的。这篇笔记不打算复述那份 PPT 里的套话而是把这类方案书里真正能落地的部分拆出来讲清楚一个园区级大数据平台从零到能用中间要过哪几道坎、参数怎么定、哪些地方最容易翻车。适合正在做农业数字化项目、手里有类似方案书但不知道怎么开工的开发和实施人员。2. 方案书里的五层架构哪几层是真需要自己写的2.1 先分清“平台能力”和“项目定制”的边界几乎所有智慧农业示范园区的方案书都会画一张分层架构图从下往上一般是感知层、传输层、数据层、服务层、应用层。看的时候容易觉得每一层都要自己建实际上要拆成两类平台能力和项目定制。平台能力指的是那些有成熟开源组件可以直接用的部分——消息队列、时序数据库、对象存储、任务调度、BI 展示。这些不需要从零写选型对了就行。项目定制指的是跟这个园区强绑定的部分——传感器协议适配、作物模型参数、园区的组织架构和权限、特定的报表口径。这部分才是真正要投入人力的地方。我一般的判断方法是拿方案书里的功能清单逐条问“换一个园区这条还能用吗”。能用的是平台能力不能用的是项目定制。一个典型的园区项目平台能力大概占 60% 的工作量但只占 30% 的代码量项目定制反过来。2.2 数据层选型时序库和关系库怎么分工农业园区的数据有个明显特点传感器数据量大但结构简单业务数据量小但关系复杂。所以数据层不要想着用一个库全吃下。传感器数据走时序数据库常见做法是 TDengine 或者 InfluxDB。TDengine 在国内农业项目里用得比较多主要原因是它对乱序写入和断网补传的容忍度好一些而且单机性能足够一个中等园区用。业务数据地块信息、种植计划、人员、设备台账走 PostgreSQL 或 MySQL这部分没什么争议。下面是一个最小可用的建表脚本用 TDengine 建传感器数据表用 PostgreSQL 建业务表-- TDengine传感器时序数据按大棚分表 CREATE DATABASE IF NOT EXISTS agri_sensor KEEP 365 DAYS 10 BLOCKS 6; USE agri_sensor; -- 超级表所有传感器共用的结构 CREATE STABLE IF NOT EXISTS sensor_data ( ts TIMESTAMP, value FLOAT, quality TINYINT ) TAGS ( greenhouse_id NCHAR(32), sensor_type NCHAR(16), sensor_id NCHAR(32) ); -- 为每个大棚创建子表写入时自动建表也可以 CREATE TABLE IF NOT EXISTS gh_001_temp USING sensor_data TAGS (GH001, temperature, T001);-- PostgreSQL业务数据 CREATE TABLE greenhouse ( id VARCHAR(32) PRIMARY KEY, name VARCHAR(64) NOT NULL, area_sqm NUMERIC(10,2), crop_type VARCHAR(32), created_at TIMESTAMP DEFAULT NOW() ); CREATE TABLE planting_plan ( id SERIAL PRIMARY KEY, greenhouse_id VARCHAR(32) REFERENCES greenhouse(id), crop_name VARCHAR(32), plant_date DATE, expected_harvest DATE, status VARCHAR(16) DEFAULT active );逻辑说明TDengine 的超级表设计是为了让同一类传感器共用 schema通过 TAGS 区分不同大棚和设备查询时按 tag 过滤效率很高。KEEP 365 DAYS 表示数据保留一年农业场景一般不需要保留太久原始秒级数据聚合后的分钟级数据可以另存。PostgreSQL 这边把大棚和种植计划分开是因为一个棚一年可能种好几茬一对多关系。参数说明BLOCKS 6控制每个 vnode 的数据块数量园区级数据量设 6 够用设太大反而增加合并开销。quality字段用来标记数据质量0 表示正常1 表示补传2 表示异常值这个字段在后期做数据清洗时非常关键一开始就要留。2.3 采集层协议适配是绕不过去的脏活方案书里通常写“支持 Modbus、MQTT、HTTP 等多种协议”实际做的时候你会发现园区里的设备协议比这杂得多。常见的有Modbus RTU 走 RS485、Modbus TCP、MQTT、还有厂商私有协议走 TCP 裸 socket。我的做法是写一个统一的采集适配层每种协议实现一个 driver 接口输出统一的数据格式。下面是一个 Python 的适配层骨架# collector/driver_base.py from abc import ABC, abstractmethod from dataclasses import dataclass from datetime import datetime dataclass class SensorReading: sensor_id: str greenhouse_id: str sensor_type: str value: float ts: datetime quality: int 0 class DriverBase(ABC): abstractmethod def connect(self) - bool: 建立连接返回是否成功 pass abstractmethod def read(self) - list[SensorReading]: 读取一批数据返回标准化后的列表 pass abstractmethod def close(self): pass# collector/modbus_driver.py from pymodbus.client import ModbusTcpClient from .driver_base import DriverBase, SensorReading from datetime import datetime class ModbusTcpDriver(DriverBase): def __init__(self, host, port, unit_id, register_map): self.client ModbusTcpClient(host, portport) self.unit_id unit_id # register_map: {sensor_id: (register_addr, scale, sensor_type)} self.register_map register_map def connect(self): return self.client.connect() def read(self): readings [] for sensor_id, (addr, scale, stype) in self.register_map.items(): # 读保持寄存器每个传感器占 1 个寄存器 resp self.client.read_holding_registers( addr, count1, slaveself.unit_id ) if resp.isError(): # 读失败标记 quality2不中断整批 readings.append(SensorReading( sensor_idsensor_id, greenhouse_id, sensor_typestype, value0.0, tsdatetime.now(), quality2 )) continue raw resp.registers[0] readings.append(SensorReading( sensor_idsensor_id, greenhouse_id, sensor_typestype, valueraw * scale, tsdatetime.now(), quality0 )) return readings def close(self): self.client.close()逻辑说明DriverBase 定义了三个方法所有协议驱动都实现这三个。read 方法返回标准化的 SensorReading 列表上层不需要知道数据是从 Modbus 还是 MQTT 来的。Modbus 驱动里对单个寄存器读取失败做了容错标记 quality2 而不是抛异常因为农业现场传感器掉线是常态不能因为一个探头坏了就停掉整个采集。参数说明scale是缩放系数Modbus 寄存器里存的是整数比如温度存的是 253 表示 25.3 度scale 就是 0.1。unit_id是 Modbus 从站地址同一个 RS485 总线上可能有多个设备每个设备一个 unit_id。register_map建议做成配置文件而不是硬编码因为不同大棚的传感器地址不一样。3. 从原始数据到能看的指标清洗和聚合怎么做3.1 农业数据的四类脏数据传感器数据接进来之后不能直接用农业场景有四类脏数据特别常见第一类是漂移值比如温度传感器慢慢偏了 2 度这种不会突变靠阈值判断抓不到。第二类是尖刺某个时刻突然跳到 80 度然后又回来通常是电磁干扰。第三类是缺失网络断了或者设备重启一段时间没有数据。第四类是重复断网补传的时候同一条数据可能传了两次。这四类里尖刺和重复最好处理漂移最难。我的经验是漂移不要试图用算法自动修而是定期人工校准算法只负责标记可疑区间。3.2 用滑动窗口做异常检测和补值下面是一个用 Pandas 做清洗的脚本处理尖刺和缺失import pandas as pd import numpy as np def clean_sensor_series(df, value_colvalue, window12, z_threshold3.5): df: 包含 ts 和 value 的 DataFrame按 ts 排序 window: 滑动窗口大小12 表示 12 个采集周期 z_threshold: 超过这个倍数中位数绝对偏差视为异常 df df.sort_values(ts).reset_index(dropTrue) series df[value_col].astype(float) # 用中位数和 MAD 做鲁棒异常检测比均值和标准差抗尖刺 rolling_median series.rolling(window, centerTrue, min_periods3).median() mad (series - rolling_median).abs().rolling( window, centerTrue, min_periods3 ).median() # 避免 mad 为 0 导致除零 mad mad.replace(0, np.nan) modified_z 0.6745 * (series - rolling_median) / mad # 标记异常 df[is_outlier] modified_z.abs() z_threshold # 异常值置为 NaN然后用线性插值补 df.loc[df[is_outlier], value_col] np.nan df[value_col] df[value_col].interpolate( methodlinear, limit6 ) # 标记补过的位置 df[quality] 0 df.loc[df[is_outlier], quality] 2 return df逻辑说明用中位数绝对偏差MAD而不是标准差是因为标准差本身会被尖刺拉大导致检测不出来。0.6745这个系数是让 MAD 在正态分布下与标准差可比。centerTrue让窗口以当前点为中心这样不会因为窗口偏向一侧而漏检。插值限制limit6表示最多补 6 个点超过就留空因为连续缺失太多说明设备真有问题补出来的值不可信。参数说明window12对应 12 个采集周期如果采集频率是 5 分钟一次就是 1 小时窗口这个尺度对温度合适对光照可能偏大。z_threshold3.5比常用的 3 更宽松因为农业数据本身波动就大设太严会把正常波动当异常。3.3 聚合层分钟级到小时级的降采样原始数据是秒级或分钟级的但看趋势的时候不需要这么细。常见做法是存三份原始数据保留 7 天分钟级聚合保留 90 天小时级聚合保留 2 年。聚合用 TDengine 的流计算或者定时任务都行。下面是一个定时聚合的 SQL-- 每小时执行一次把上一小时的分钟数据聚合成小时数据 INSERT INTO agri_sensor.hourly_avg SELECT _wstart AS ts, AVG(value) AS value, AVG(quality) AS quality, greenhouse_id, sensor_type, sensor_id FROM agri_sensor.sensor_data WHERE ts NOW - 1h AND ts NOW INTERVAL(1h) GROUP BY greenhouse_id, sensor_type, sensor_id;逻辑说明TDengine 的INTERVAL子句做时间窗口聚合_wstart是窗口起始时间。这里用 AVG 而不是 LAST因为小时均值更能反映趋势。quality 也取平均如果一小时里超过一半的点是异常均值会大于 1后期查询时可以据此过滤。参数说明INTERVAL(1h)可以改成INTERVAL(10m)做更细的聚合。聚合表建议单独建不要和原始表混在一起因为查询模式完全不同。4. 避坑园区大数据平台最常见的五个翻车点4.1 坑一传感器时间戳不可信现象数据写进去之后按时间查出来的曲线是乱的有时候一个点跑到未来去了。原因很多低成本传感器模块没有 RTC上电后时间从 2000 年 1 月 1 日开始走或者网关的时间没同步。数据到了平台这边如果直接用设备上报的时间戳就会乱。解决平台侧统一用接收时间作为主时间戳设备时间只作为参考字段存下来。如果确实需要设备时间在网关上做 NTP 同步并且平台侧做一次合理性校验——时间戳跟服务器时间差超过 5 分钟的直接标记 quality2。4.2 坑二断网补传导致数据重复现象某个时间段的数据量突然翻倍聚合出来的均值明显偏低或偏高。原因网关断网时本地缓存网络恢复后批量补传但补传逻辑没做去重或者平台侧没做幂等。解决写入时用(sensor_id, ts)做唯一约束TDengine 里可以用INSERT IGNORE或者建表时指定主键。更稳妥的做法是在采集层给每条数据加一个序列号平台侧记录每个传感器已处理的最大序列号小于等于的直接丢弃。4.3 坑三时序库磁盘写满现象平台跑了两三个月突然写不进去了查日志发现磁盘 100%。原因原始数据没设保留策略或者保留策略设了但没生效。TDengine 的 KEEP 参数是在建库时指定的后期改需要ALTER DATABASE。解决建库时就把 KEEP 设好并且配一个监控告警磁盘到 80% 就通知。另外聚合任务要确保在跑否则原始数据删了之后没有降采样数据可查。4.4 坑四可视化查询拖垮数据库现象大屏打开特别慢有时候直接超时但数据库 CPU 并不高。原因大屏的查询没走聚合表直接查原始表做 GROUP BY数据量大时扫描行数太多。或者多个大屏同时刷新并发查询把连接池占满。解决所有大屏查询强制走聚合表原始表只给排查问题用。查询结果加缓存比如 Redis 缓存 5 分钟。连接池大小根据数据库配置调整TDengine 默认连接数不多要提前设好。4.5 坑五权限模型没设计好后期改不动现象一开始只有管理员和普通用户两种角色后来园区要求按大棚分权限发现改起来要动所有接口。原因权限判断散落在各个业务代码里没有统一的鉴权层。解决一开始就把权限模型设计成用户-角色-资源三层资源粒度到单个大棚。接口层统一用一个装饰器做鉴权业务代码里不写权限判断。这个改动在项目初期成本很低后期补代价很大。5. 让平台真正被用起来的一个小技巧平台建好之后最大的风险不是技术问题是没人用。我见过好几个园区大屏挂在墙上数据也在更新但种植人员还是凭经验浇水施肥平台成了一个展示品。让平台被用起来我的经验是找一个最小闭环先跑通不要一上来就做全功能。最小闭环的意思是从一个传感器到一个动作中间不经过人工判断。比如土壤湿度传感器读数低于阈值 → 自动发一条微信消息给负责该大棚的人 → 消息里带一个“已处理”按钮 → 点击后记录处理时间连续 3 天温度低于作物生长下限 → 自动生成一条预警工单 → 工单派给对应负责人这个闭环里没有复杂的数据分析但它是可验证的——你能看到消息有没有发出去、有没有人点、处理了多久。跑通一个闭环之后再往上加指标、加报表、加模型种植人员才会觉得这个平台跟他的日常工作有关。下面是一个最小闭环的伪代码用定时任务实现# tasks/alert_task.py import requests from datetime import datetime, timedelta THRESHOLDS { soil_moisture: {min: 30, max: 80, unit: %}, temperature: {min: 10, max: 35, unit: C}, } def check_and_alert(db, notifier): 每 10 分钟跑一次检查最近 30 分钟的数据 since datetime.now() - timedelta(minutes30) for sensor_type, limits in THRESHOLDS.items(): # 查最近 30 分钟的均值 rows db.query( SELECT greenhouse_id, AVG(value) AS avg_val FROM sensor_data WHERE sensor_type %s AND ts %s AND quality 0 GROUP BY greenhouse_id , (sensor_type, since)) for row in rows: val row[avg_val] if val limits[min] or val limits[max]: # 检查是否已经发过告警避免重复 if not db.has_open_alert(row[greenhouse_id], sensor_type): notifier.send( titlef{row[greenhouse_id]} {sensor_type} 异常, contentf当前值 {val:.1f}{limits[unit]} f正常范围 {limits[min]}-{limits[max]}, callback_data{ greenhouse_id: row[greenhouse_id], sensor_type: sensor_type, } ) db.create_alert(row[greenhouse_id], sensor_type, val)逻辑说明这个任务每 10 分钟跑一次查最近 30 分钟的均值而不是瞬时值避免因为一个尖刺就发告警。has_open_alert检查是否已经有未处理的告警防止重复发送。callback_data用来在用户点击“已处理”时回写状态。参数说明THRESHOLDS里的阈值应该做成配置不同作物、不同生长阶段阈值不一样。since取 30 分钟是因为传感器采集周期通常是 5 分钟30 分钟有 6 个点均值比较稳定。我自己的习惯是每做一个园区项目先花两天时间把最小闭环跑通哪怕只接一个传感器。这两天里能暴露出来的问题——网络、供电、协议、时间戳——比后面做一个月报表都多。等闭环跑通了再按方案书里的架构往上堆心里就有底了。希望帮到你。本文还有配套的精品资源点击获取