10.3 工业时序数据与规则引擎
10.3.1 时序数据库选型与数据模型
数据从边缘网关汇聚到平台层之后，第一个需要解决的问题是：用什么来存？
工业场景下的数据流有自己的脾气。一台数控机床的振动传感器每秒上报上千个采样点，一条产线上百个温度探头每两秒一个位号，这些数据合在一起按年计，写压力容易破千万甚至上亿点/天。更关键的是，这些值天然自带时间戳——这就是时序数据的核心特征。
关系型数据库与专用时序数据库各有边界。PostgreSQL 通过分区、批写、合适索引和扩展也能承载大量时序数据；专用 TSDB 则可能在压缩、保留和时间聚合上提供更直接的能力。是否“不划算”只能由目标写入、查询、保留、事务和运维条件下的基准决定，不能把产品类别写成性能结论。
时序数据库的核心特点
时序数据库为工业数据流做的专门设计，可以概括为四条：LSM-Tree（Log-Structured Merge-Tree）类结构把随机写入转化为顺序追加，换取高写入吞吐；按时间窗口自动切分分区，查询只扫相关分区；降采样与聚合运算下推到存储层执行；按保留策略（Retention Policy）自动过期清理分区。这些机制的引擎级原理——写入路径、压缩编码、持续聚合与冷热分层——第 5 章 5.4 节已逐一拆解过，本节不再重讲，只回答工业项目里更常纠结的那一问：具体选哪一款。
主流时序数据库选型
工业物联网平台面临的选型不是“用不用时序库”，而是“用哪种”。几款主流产品在工业场景的能力边界不同。
表10-4：主流工业时序数据库特性对比
| 特性维度 | InfluxDB (1.x / 3.x) | TimescaleDB | TDengine |
|---|---|---|---|
| 架构类型 | 独立TSDB引擎（自主研发存储） | PostgreSQL扩展 | 独立TSDB引擎（自主研发存储） |
| 数据模型 | 度量（measurement） + 标签（tags） + 字段（fields） | 超表（hypertable） + 列 | 超级表（supertable） + 标签 + 列 |
| 写入性能 | 取决于版本、Schema、批量、硬件和持久化设置，需实测 | 取决于 PostgreSQL 配置、分区、索引和批写，需实测 | 取决于版本、表模型、硬件和副本设置，需实测 |
| SQL兼容性 | 自定义InfluxQL/Flux | 完整PostgreSQL SQL | 类SQL（支持有限Join/窗口函数） |
| 集群与高可用 | 1.x开源版无集群；3.x支持集群 | 基于PG流复制，需自行搭建 | 企业版支持；开源版无原生集群 |
| 适用场景 | 中小规模监控、运维监控、IoT平台 | 需要复杂SQL分析、与PG生态结合的产线 | 高吞吐、高压缩率的大规模工业位号 |
选型没有绝对答案。需要说明的是，InfluxDB 的 2.x（引入 Flux 与 TSM 重构的版本）在官方路线中被视为过渡版本，当前主线是 1.x 与 3.x，因此本表只对比这两个系列。如果团队已经重度依赖 PostGIS 和复杂业务查询，TimescaleDB 能复用现有 SQL 技能栈；如果场景是单一的“传感器写→监控看→告警”，InfluxDB 更轻量；如果年数据量在数十亿点以上且要求高压缩率，TDengine 的列式存储选型值得评估。
IoT DC3 在设计时并没有锁死某一种时序库，而是通过数据中心层抽象了存储接口，允许在生产环境中按需切换底层的时序存储引擎（TimescaleDB、TDengine 等）。
位号与标签设计：数据模型的关键
时序数据库的威力不仅依赖存储引擎，更依赖数据模型的合理设计。在 IoT DC3 的实践中，一条时序数据被建模为 PointValue —— 每个值都带五个固定属性：
- device_id（设备ID）：关联物理设备实例。
- point_id（位号ID）：唯一标识一个传感器或寄存器地址。
- value（数值/状态）：经过归一化处理的实际工程值。
- event_time（采集时间戳）：设备端或网关端打标的时间。
- unit（单位）：单位上下文（如℃、kPa、rpm），用于语义解析。
除此之外，标签（Tag） 是可选的维度字段，用于支持多维查询——例如通过“产线=产线A AND 工序=焊接”来检索所有与该工序相关的温度位号。
-- 示意：IoT DC3 基于 TimescaleDB 的时序表结构
CREATE TABLE point_value (
device_id VARCHAR(64) NOT NULL,
point_id VARCHAR(64) NOT NULL,
event_time TIMESTAMPTZ NOT NULL,
value DOUBLE PRECISION NOT NULL,
unit VARCHAR(16),
quality SMALLINT DEFAULT 1, -- 0=已弃质, 1=正常
-- 可选：标签列（通过物模型预定义）
tags JSONB DEFAULT '{}'::jsonb,
PRIMARY KEY (device_id, point_id, event_time)
);
-- 按设备和时间做分区（Hypertable）
SELECT create_hypertable('point_value', 'event_time', chunk_time_interval => INTERVAL '1 day');
-- 按设备ID做空间分区分流
SELECT add_dimension('point_value', 'device_id', number_partitions => 16);数据模型设计阶段最容易踩的坑有两个。
一、标签基数爆炸。 把每一条数据都带上“产线、工序、设备型号、制造商、批次号”等大量标签，虽然查询灵活，但会导致时序库的倒排索引膨胀到不可控。工业上一条产线几百个位号，每个位号带六七个标签，索引体积可能超过数据本身的数倍。建议主维度的标签控制在三到五个内，其余维度通过外键关联到元数据表查询，不要全部塞进时序表。
二、不分主副的时间分区。 同一台设备的振动和温度采样频率可能差两个数量级。如果强行统一时间分区，低频数据的分区存储浪费严重。比较好的做法是按位号类型分表或分分区键：高频振动走短时间窗口（如每小时分区），低频温度走长窗口（如每天分组）。
数据模型的选择还直接决定了后续规则引擎和 AI 模型的消费成本。好的模型在设备连接侧已经完成了“tag 用于过滤、value 用于计算、time 用于对齐”的分工；坏的模型则把麻烦全推给数据处理层——在大幅增加查询复杂度的同时，进一步加大了系统延迟。
设计时序数据模型时，建议在一个检查清单上逐个确认：
- 每个 point_id 是否在物模型中定义了明确的语义（物理含义 + 数据类型 + 单位）？
- 标签（tags）的基数和可能取值是否已预先评估？
- 是否按采样频率差异做了分区策略？
- 数据保留策略如何设定——原始数据保留多久？降采样如何执行？
- 写并发最高的时刻是什么？峰值写入速率是否经过压测验证？
这一节集中在数据模型层面。有了干净、可查询的时序数据，下一步是让这些数据动起来——被规则引擎消费，触发告警或自动决策。这正是 10.3.2 要展开的内容。
10.3.2 规则引擎原理与工业告警设计
时序数据库把数据落盘，解决了“存得住”的问题。但工业场景的真正价值在于“反应快”：设备温度越过阈值要立刻告警，振动值连续异常要触发停机流程，多参数联合判断需要把温度、压力、电流放在一条规则里综合评估。这层逻辑如果写死在应用代码里，改一条阈值就要重新部署，不可接受。规则引擎的价值正在于此：把“判断—执行”从业务代码中抽离，变成可配置、可热更新的规则集。
事件驱动与条件判断
工业告警的输入通常是一条时序位号数据流。规则引擎以事件驱动（Event-Driven） 的方式运行：每一条新上报的位号值都作为一条事件推入引擎的推理工作内存（Working Memory）。引擎采用改进的Rete算法实现高效模式匹配——它将规则的条件编译成网络结构，通过增量匹配避免每次触发都重算全部事实。Rete的优势在规则数超过百条时尤为明显；若规则仅几十条，简单线性扫描也可接受，选型时不必过度设计。
以IoT DC3平台为例，规则引擎模块接收来自数据中心的PointValue（带语义标签、单位、时间戳的归一化位号值）。工程师可以在规则中心编写规则，例如“电机1号轴承温度 > 85℃ 且持续时间超过10秒”。规则引擎每收到一个温度位号值，即开始条件评估，并在窗口闭合时触发动作。
以下是一个规则定义片段，展示了条件判断与动作执行的配置：
{
"ruleId": "bearing-temp-high-001",
"name": "电机1号轴承温度过高",
"description": "检测电机1号轴承温度持续超过85℃",
"priority": 10,
"condition": {
"type": "continuous",
"measurement": "temperature",
"deviceId": "motor-01",
"pointId": "bearing-temp",
"operator": ">",
"threshold": 85,
"durationSeconds": 10
},
"action": {
"type": "alarm",
"severity": "critical",
"notify": ["sms", "email"],
"hookUrl": "http://alert-service/api/v1/alarms"
},
"enabled": true
}该配置的语义：当设备motor-01的bearing-temp位号值在10秒内持续高于85时，触发一条严重级别为critical的告警，通过短信和邮件通知，并调用外部告警服务的REST接口。规则权重priority:10决定了它在冲突集中的执行优先级——值越高越先执行。需要说明的是，这是一个工程例子，实际生产环境中的规则定义会根据平台和协议有所调整，但核心结构类似。
规则优先级与冲突解决
当多条规则同时满足条件时（例如温度超高告警和振动异常告警同时触发），引擎需要决定先执行哪一条。Drools等主流规则引擎把满足条件的候选执行项放在议程（Agenda）上，按冲突解决策略（Conflict Resolution）排序执行，默认排序主要看两条：
- 优先设置（Salience）：工程师为每条规则显式指定一个整数值，值越高，执行优先级越高。这是最常用的手段。紧急告警规则通常分配较高值以确保它先于非紧急规则执行。在不指定时，默认值为0。
- 激活新近度（Recency）：salience相同时，越晚被激活的规则越先执行（类似栈的后进先出）。对工业告警而言这是个合理的默认——同一条规则被连续触发时，携带最新事实的激活会先得到处理。
- 议程分组（Agenda Group）：将规则归为不同分组，引擎按组顺序执行。适用于按流程阶段划分的场景，例如先执行“数据质量检测”组，再执行“工况判断”组。同一分组内仍需靠优先设置排序。
需要澄清一个常见误传：“引擎默认仅激活条件更具体的规则”是CLIPS等引擎的可选策略（specificity），并非Drools的默认行为——Drools默认就是salience加激活新近度。因此两条条件重叠的规则（例如temperature > 90和temperature > 85同时满足）默认都会被激活、先后执行，重复通知要靠工程师自己消除：常见做法是让具体规则以更高的salience覆盖一般规则，或依靠告警抑制窗口合并同源告警（见本节后文）。
工程上的一个常见陷阱是：过度依赖优先设置而不分组，导致规则数量增多后排序混乱。建议在规则数超过50条时引入议程分组，按业务阶段（如数据质量→工况判断→告警产生→工单创建）切分，每组内再控制不超过10条规则。
告警分级与通知渠道
告警在工业现场不是一件事——它是一个层层递进的操作流程。一般使用三级分级（工程惯例，非标准强制）：
- 提醒（Info）：阈值接近但未超限。通知方式：日志记录、监控看板标签，无需主动推送。
- 预警（Warning）：阈值超限但仍在安全边界内，设备仍可运行。通知方式：工单系统、邮件、看板闪烁。
- 紧急（Critical）：阈值超限且影响设备安全或会引发连锁停线。通知方式：短信、电话语音告警或MES系统自动停机指令。
通知渠道的选择取决于响应时间要求。一个合理的分级结构如下：
| 告警级别 | 响应时间要求 | 推荐通知渠道 | 是否需要工单 |
|---|---|---|---|
| Critical | 数分钟内 | 短信 + 电话 + MES接口 | 是 |
| Warning | 几小时内 | 邮件 + 看板 | 是 |
| Info | 日常巡检 | 看板 + 日志 | 否 |
拆分通道不是为了“功能丰富”，而是为了降低运维噪音。把所有阈值超限都用短信推一次的结果，是运维人员对短信麻木，错过真正的紧急事件。一个务实的工程判断是，让Info级别的规则在数量上占据主体，Critical级别严格控制，以避免告警疲劳。同时，应设置告警抑制：同一设备同一类告警在设定时间窗口（例如30分钟）内只触发一次，除非状况升级。
规则引擎状态迁移
规则引擎在运行中并非只有“激活—执行”两种状态。设计合理的规则引擎应具备以下状态迁移能力：规则从DRAFT（草稿）创建，经手动启用进入ENABLED（激活），收到匹配事件后进入MATCHED（匹配），被引擎选中执行后进入EXECUTED（已执行），执行后事实更新重置回到ENABLED。规则也可以从ENABLED或DRAFT手动进入DISABLED（停用），最终进入DELETED（删除）。需要说明的是，MATCHED/EXECUTED这组运行态是IoT DC3规则中心自定义的状态模型，用于描述本书示例中的规则生命周期，并非Drools等通用规则引擎的标准语义——通用引擎里的对应概念是议程上的激活（Activation）与点火（Fire）。这个状态机设计的核心价值在于热更新：规则不需要重启服务就能从DRAFT进入ENABLED，从DISABLED恢复。在产线不停机的前提下修改告警阈值，正是工业场景对规则引擎的硬性要求。实际落地时需注意：从ENABLED到MATCHED的迁移依赖于工作内存中的事实——如果历史数据未被清空，新加的规则可能瞬间匹配到已过期的事实，产生误警。因此建议在启用规则时清空对应设备的旧事实，或让规则条件附加timestamp > now - 5s这类时间约束。
规则到模型的过渡边界
规则引擎擅长处理明确的、可枚举的条件判断。但当判断条件从“温度>85”变为依赖振动频谱特征、需要结合历史故障模式做模式识别时，规则配置就难以胜任了——阈值变得模糊，依赖历史数据和特征提取。这时应该把规则引擎视为一个触发层，将分析推理交给训练好的AI模型：规则引擎根据检测到的基础特征（如有效值超过基线）调用REST接口将特征数据传递给推理服务，后者返回故障概率，规则引擎再根据概率阈值生成相应级别的告警。10.4节会展开这条“规则+模型”的混合链路。
在部署规则引擎前，建议先遍历产线上每种设备的告警场景，用以下检查清单判断：“哪些适合写死阈值、哪些需要时间窗口、哪些必须借助历史数据”。分清楚之后，大部分场景可以落在规则引擎的射程内，剩余部分留给模型接入。这个划分依据的是工程经验，用于指导任务切分，而非精确统计。
规则引擎工程检查清单（投入产线前必检）：
- [ ] 每条规则是否设置了明确的优先级（Salience）和分组（Agenda Group）？
- [ ] 告警分级的通知渠道是否与响应时间要求匹配，有无过度推送？
- [ ] 是否配置了告警抑制：同一设备同一类告警在设定窗口内只触发一次？
- [ ] 规则热更新是否经过测试（从DRAFT切换为ENABLED后，旧事实是否已清理）？
- [ ] 规则执行性能：规则数量上限和Rete网络深度是否已在开发环境压力测试？
- [ ] 是否预留了模型层的REST接口，以便将来从固定阈值升级为概率判断？
10.3.3 数据质量与异常值处理
时序数据和规则引擎构成的“感知-判断”链路中，输入质量决定输出效果。工业现场的数据采集并非理想环境：传感器老化、通信干扰、PLC缓存溢出、网关断连，都会导致数据出现缺失、毛刺、重复。这类问题如果不处理就推送进规则引擎或AI模型，结果多是误报或漏报，且难以事后追溯。
但工业数据质量治理第一步不是“清洗”，而是标记。在IoT DC3这类平台中，每条位号值都带有时间戳和状态字段（如 quality 标记），可用来区分“正常”“可疑”“坏值”。清洗策略应作用于已标记的数据，而非盲目修改原始记录。
缺失数据处理
工业时序缺失可能来自传感器故障、网络中断、停机或采集任务变更，先判原因，再决定是否插值。连续点数不是通用门槛：同样缺 3 个点，对毫秒级振动与小时级储罐温度含义完全不同。前向填充和线性插值只能生成分析用衍生序列，必须保留原始缺口、质量码、方法与最大插值时长；控制、安全联锁和事故取证不得把插值值冒充实测值。
毛刺过滤
毛刺表现为单点或连续几个点大幅偏离正常范围，俗称“尖峰”。工程上常用基于中位数的滑动窗口过滤：设定窗口长度（如5点），计算窗口内中位数，若当前值与中位数的绝对差超过预设阈值（例如以正常运行标准差的三倍为门限），则判定为毛刺。替换值可选择中位数或窗口均值。阈值设定必须考虑设备工况：正常启停机时的剧烈变化不应被视为毛刺。
重复数据去重
重复数据通常由网关或协议冗余上报引起。最简单的做法是以设备ID加时间戳为唯一键，在接收端做幂等处理。时序数据库本身通常支持按时间戳去重，但需设计冲突解决策略：若两条相同时间戳的数据值不同，有两种常用方案——保留最新时间戳的数据，或标记为“冲突”并由人工判断。
示例代码
以下是一段Python清洗代码，展示缺失填充、毛刺过滤和去重的基本操作。代码中的abs_dev是当前值距滑动中位数的绝对偏差，注意它不是统计学里的标准MAD（median absolute deviation，中位绝对偏差，定义为median(|x−median(x)|)，对整段窗口再取一次中位数）——标准MAD稳健性更强，但需要逐窗计算，示意实现选择了更轻的折中。生产环境中此逻辑一般放在边缘网关或平台预处理阶段，且必须参考设备工艺参数微调阈值。
import pandas as pd
import numpy as np
# 假设df为温度序列，列'value'，时间戳索引
# Step 1：仅生成分析副本；limit 必须由过程动态和采样周期验证
df['value_filled'] = df['value'].ffill(limit=validated_gap_limit)
df['is_imputed'] = df['value'].isna() & df['value_filled'].notna()
# Step 2：基于中位数的滑动窗口毛刺过滤（窗口=5）
window = 5
df['median'] = df['value_filled'].rolling(window, center=True).median()
df['abs_dev'] = np.abs(df['value_filled'] - df['median'])
# 此处使用距滑动中位数的绝对偏差均值的3倍作为示意阈值，实际需根据工况校准
threshold = 3 * df['abs_dev'].rolling(window, center=True).mean()
mask = df['abs_dev'] > threshold
df['value_clean'] = np.where(mask, df['median'], df['value_filled'])
# Step 3：按时间戳去重（保留第一个值，适用多数以帧为单位上报的场景）
df = df[~df.index.duplicated(keep='first')]这些清洗策略不能解决所有问题。数据质量长期低下时，应先排查设备或通信链路，而不是依赖算法修补。项目应定义可量化的质量指标，并明确其计算口径和责任人。清洗与质量标记后的数据可供诊断 Agent 通过受控 Tool 查询；MCP 只负责暴露工具，不替平台下发设备指令，也不保证模型判断正确。