EQL — 事件查询语言帮助
目录
- 概述
- 在哪里使用 EQL
- 基本语法
- 事件流 (14 种 + 内部用 3 种)
- 窗口
- WHERE 条件
- 聚合函数
- 分组·排序
- JOIN — 流结合
- 模式匹配
- 时间·状态函数
- 变量与上下文
- 12 个实战案例
- 性能指南
- 常见问题
- 相关界面
概述
主要使用位置：在左侧菜单 自动化 > EQL 查询(
/query/index) 中直接运行，在 高级报警 · 触发器·语句 · 流 的条件表达式中也使用相同的语法。
EQL (Event Query Language) 是 PlantPulse Platform 为了能够像 SQL 语法一样对实时流数据进行查询、聚合和模式匹配而开发的自有查询语言。如果说通用 SQL 是针对静态表运行的，那么 EQL 则在持续流入的事件流之上运行，并实时发射 (emit) 结果。
运维人员可以通过一行 EQL 实现以下功能：
- 实时监控 — 标签值超过阈值的瞬间立即发射结果
- 资产单位采集 — 每 5 分钟自动计算平均值/最大值/最小值
- 模式匹配 — “A 事件发生后 30 秒内如果没有 B 事件则报警”
- 时序连接 — 比较两个标签的时间对齐情况
EQL 使用与 SQL 几乎相同的语法，但增加了 窗口 (Window) 概念，用以指定如“最近 N 秒 / N 条”等时间或长度范围。
本页面涵盖了在控制台中运行的所有 EQL。但 CEP 在启动时会选择一个后端，每个后端支持的配置有所不同 — 某些功能（如连接、变量、上下文）仅在部分后端中可用。
具体哪些功能在哪个后端运行，请参阅 CEP — 后端 表格；语言本身请参考 CEP — EQL 语言；拒绝令牌请参考 限制与拒绝令牌。
拒绝令牌在注册时就会提示名称，因此不会在运行时静默失败。
在哪里使用 EQL
Platform 内 EQL 的使用位置如下：
EQL 查询 · 语句位于左侧菜单 자동화 组中。 没有名为 CEP 的菜单。
| 菜单 · 界面 | 内部 URL | 用途 |
|---|---|---|
| 自动化 > EQL 查询 — 文档 | /query/index | 运维人员即时运行 EQL 并确认结果（探索·调试） |
| 触发器 (左侧菜单无) — 文档 | /trigger/index | 将 EQL 匹配结果自动发布到 MQ 主题或存储 |
| 自动化 > 语句 — 文档 | /statement/index | 按资产单位注册的规则（状态/采集/事件/命令）。必须填写各类型 必填输出项 才能保存 |
| 报警 → EQL 报警 | — | 通过 EQL 规则发布任意条件报警 |
| 流 → 脚本 | — | flow_script_filter / flow_script_transform / flow_switch 的 EQL 表达式模式 |
| 标签样式 — 人工智能页签 | — | 预测·异常检测规则的输入条件 |
EQL 是只读的 — 它不会修改数据。若要通过匹配结果发布报警或调用 流 动作节点，请通过 CEP → 触发器 或 EQL 报警 连接结果。
基本语法
EQL 的基本结构与 SQL 几乎相同。
SELECT <column-or-function>
FROM <stream>.<window>
WHERE <condition>
GROUP BY <field>
HAVING <aggregation-condition>
ORDER BY <field> [ASC|DESC]
最简单的 EQL：
SELECT * FROM Point.win:time(1 sec)
每秒发射一次最近 1 秒内流入的所有标签数据点。
通配符与列选择
| 语法 | 含义 |
|---|---|
SELECT * | 包含所有列 |
SELECT tag_id, value | 仅特定列 |
SELECT value AS v, tag_id AS id | 赋予别名 (alias) |
SELECT value * 9 / 5 + 32 AS fahrenheit | 算术表达式结果 |
SELECT CASE WHEN value > 80 THEN 'HIGH' ELSE 'OK' END AS level | 条件表达式 |
注释
-- 한 줄 주석
/* 여러 줄
주석 */
SELECT * FROM Point.win:time(1 sec)
事件流 (14 种 + 内部用 3 种)
EQL 的 FROM 子句中可使用的流 (事件源) 如下。所有流均由 Platform 自动产生，无需单独注册即可立即使用。
如果在
FROM中使用此处未列出的名称，规则将无法部署。只有注册的事件类型才能成为流 (plantpulse.core.engine.eql.EventTypes)。
标签域 (3 种)
| 流 | 发生时点 | 主要字段 |
|---|---|---|
| Point | 每次接收到标签值时 | tag_id, value, quality, ts, site_id, asset_id, unit, java_type |
| PointMap | 同上，但与 tag('<id>') 函数配合使用，以键-值形式查询多个标签 | (PointMap 专用助手) |
| Alarm | 标签报警发生时 | tag_id, alarm_band, value, threshold, priority, band_message |
资产域 (6 种)
| 流 | 发生时点 | 主要字段 |
|---|---|---|
| AssetData | 资产单位时序数据 (采集结果) | asset_id, values (key-value), ts |
| AssetEvent | 资产事件发生 (启动·停止·维护等) | asset_id, event_type, details, ts |
| AssetAlarm | 资产报警发生 | asset_id, alarm_band, priority, message |
| AssetCommand | 向资产发布命令时 | asset_id, command, args |
| AssetAggregation | 1 分钟/5 分钟/1 小时资产采集结果 | asset_id, window, agg_type, values |
| AssetContext | 资产上下文 (元数据) 变更 | asset_id, context_key, before, after |
旧版文档中将这两个列为资产域流，但它们不是注册的事件类型。
如果写入 FROM AssetHealthStatus...，规则将无法部署。
健康·连接状态评估是通过周期执行作业运行的，而非流
(AssetHealthStatusDeployer / AssetConnectionStatusDeployer)。其结果通过界面和 API 报告。如果在 EQL 中想要捕捉状态变化，请通过 not 模式 检测源端的 AssetEvent · AssetAlarm 或 Point 的未接收情况。
生产域 (3 种)
| 流 | 发生时点 | 主要字段 |
|---|---|---|
| Calendar | 班次开始/结束/变更 | shift_id, calendar_id, event, start_time, end_time |
| WorkOrder | 工单生命周期转移 | order_id, status, prev_status, asset_id, product_id |
| OEE | OEE 评估结果更新 | order_id, asset_id, oee, availability, performance, quality |
系统域 (2 种)
| 流 | 发生时点 | 主要字段 |
|---|---|---|
| Log | 系统诊断消息 | module, level, code, summary, ts |
| OSPerformance | 服务器 OS 资源指标 (周期性) | CPU·内存·磁盘等 |
不存在名为 Status 的事件类型。如果想通过 EQL 捕捉 OPC·Edge 的连接状态变化，不能使用该流 — 请使用 «not 模式» 检测 Point 未接收的方式。
内部用流 (3 种)
虽然已注册，但不用于编写规则。用于诊断·健康检查。
| 流 | 用途 |
|---|---|
| DateTime | 周期性发生的时刻事件 (时间基准规则的触发器) |
| Ping | 引擎生存确认 |
| Test | 规则验证用 |
所有流在使用前 必须与窗口结合 (
Point.win:time(1 sec))。如果仅使用FROM Point而不带窗口，结果会立即消失而无法查看。
窗口
窗口是“从流中截取的时间或长度区间”。EQL 支持两种窗口。
时间窗口 (Time Window)
| 语法 | 含义 |
|---|---|
Point.win:time(1 sec) | 最近 1 秒 内流入的所有事件 (sliding) |
Point.win:time(10 sec) | 最近 10 秒 |
Point.win:time(5 min) | 最近 5 分钟 |
Point.win:time(1 hour) | 最近 1 小时 |
支持的时间单位：sec (秒) · min (分) · hour (时) · day (日)。
长度窗口 (Length Window)
| 语法 | 含义 |
|---|---|
Point.win:length(100) | 最近 100 条 事件 |
Point.win:length(1000) | 最近 1,000 条 |
批次窗口 (Batch Window)
win:time_batch / win:length_batch — 仅在窗口填满或时间过期时刻发射一次。可节省资源。
-- 매 1분마다 한 번씩 그 1분간 평균을 emit
SELECT avg(value) FROM Point.win:time_batch(1 min) WHERE tag_id = 'MOTOR-001.TEMP'
窗口选择指南
| 场景 | 推荐窗口 |
|---|---|
| 实时监控 (查看值本身) | win:time(1 sec) |
| 短期平均/最大值 (如 5 分钟平均) | win:time(5 min) |
| 小时单位采集 | win:time_batch(1 hour) |
| 最近 N 条对比 | win:length(N) |
| 变化量跟踪 | win:length(2) + prev() 函数 |
WHERE 条件
使用与 SQL 相同的比较运算符。
| 运算符 | 示例 | 含义 |
|---|---|---|
= != <> | tag_id = 'MOTOR-001' | 相等/不等 |
< <= > >= | value > 80 | 比较 |
BETWEEN | value BETWEEN 70 AND 90 | 范围 |
IN | tag_id IN ('A','B','C') | 包含在列表中 |
LIKE | tag_id LIKE 'MOTOR-%' | 模式匹配 (% 通配符) |
IS NULL / IS NOT NULL | quality IS NOT NULL | 是否为 null |
AND OR NOT | value > 80 AND quality = 'GOOD' | 逻辑结合 |
正则表达式
SELECT * FROM Point.win:time(1 sec)
WHERE tag_id REGEXP '.*\\.TEMP$'
仅匹配所有以
.TEMP结尾的标签。反斜杠需输入两次 (\\.)。
动态站点/资产过滤器
SELECT * FROM Point.win:time(1 sec)
WHERE site_id = 'SITE-A'
AND asset_id LIKE 'LINE-1.%'
AND value > 75
聚合函数
对窗口内的事件进行采集。
基础采集
| 函数 | 含义 | 示例 |
|---|---|---|
count(*) | 事件条数 | SELECT count(*) FROM Point.win:time(10 sec) |
sum(field) | 总和 | sum(value) |
avg(field) | 平均值 | avg(value) |
min(field) | 最小值 | min(value) |
max(field) | 最大值 | max(value) |
median(field) | 中位数 | median(value) |
stddev(field) | 标准差 | stddev(value) |
variance(field) | 方差 | variance(value) |
first(field) | 窗口首值 | first(value) |
last(field) | 窗口末值 | last(value) |
DISTINCT 采集
SELECT count(distinct tag_id) AS unique_tags
FROM Point.win:time(1 min)
与 CASE 结合
SELECT
count(CASE WHEN value > 80 THEN 1 END) AS high_count,
count(CASE WHEN value <= 80 THEN 1 END) AS ok_count
FROM Point.win:time(1 min)
WHERE tag_id = 'MOTOR-001.TEMP'
时序函数
| 函数 | 含义 | 示例 |
|---|---|---|
prev(N, field) | N 步之前的值 | value - prev(1, value) (与前值之差) |
prevwindow(field) | 窗口内所有前值数组 | prevwindow(value) |
rate(field) | 单位时间变化率 | rate(value) |
分组·排序
GROUP BY
-- 자산별 최근 1분 평균 온도
SELECT asset_id, avg(value) AS avg_temp
FROM Point.win:time(1 min)
WHERE tag_id LIKE '%.TEMP'
GROUP BY asset_id
HAVING — 采集后过滤
-- 평균 > 80 인 자산만
SELECT asset_id, avg(value) AS avg_temp
FROM Point.win:time(1 min)
WHERE tag_id LIKE '%.TEMP'
GROUP BY asset_id
HAVING avg(value) > 80
ORDER BY · LIMIT
-- 가장 뜨거운 5개 자산
SELECT asset_id, max(value) AS peak
FROM Point.win:time(5 min)
WHERE tag_id LIKE '%.TEMP'
GROUP BY asset_id
ORDER BY peak DESC
LIMIT 5
LIMIT仅包含发射结果时刻最顶部的 N 条记录。
JOIN — 流结合
可以将两个或多个流按时间对齐进行连接。
-- 같은 자산에서 같은 시간대의 온도와 압력을 한 행으로
SELECT p1.value AS temp, p2.value AS pressure, p1.asset_id
FROM Point.win:time(1 sec) AS p1,
Point.win:time(1 sec) AS p2
WHERE p1.tag_id LIKE '%.TEMP'
AND p2.tag_id LIKE '%.PRESSURE'
AND p1.asset_id = p2.asset_id
资产事件 + 工单连接
SELECT e.asset_id, e.event_type, w.order_id, w.status
FROM AssetEvent.win:time(5 min) AS e,
WorkOrder.win:time(5 min) AS w
WHERE e.asset_id = w.asset_id
AND e.event_type = 'STARTUP'
AND w.status = 'START'
仅在两个流的时间窗口重叠期间产生连接结果。比较时间单位不同的数据时，请使用
win:time_batch将其对齐到相似周期。
模式匹配
EQL 的强大功能 — 可以定义如“事件 A 之后 N 秒内 B (未) 发生”等序列。
every — 每次匹配
SELECT * FROM pattern [
every a = AssetEvent(event_type = 'STARTUP')
-> b = AssetEvent(event_type = 'SHUTDOWN', asset_id = a.asset_id)
where timer:within(60 sec)
]
每次匹配 开始事件 → 60 秒内停止事件 的组合。用于探测短时间运行后立即停止的异常模式。
not — N 时间内未发生
SELECT * FROM pattern [
every a = AssetEvent(event_type = 'STARTUP')
-> ( timer:interval(30 min)
and not AssetData(asset_id = a.asset_id) )
]
开始后 30 分钟内如果没有数据流入则匹配 (探测无响应资产)。
and / or — 同时或其中之一
SELECT * FROM pattern [
( every AssetAlarm(priority = 'ERROR') )
and
( every WorkOrder(status = 'START') )
]
模式匹配核心关键字
| 关键字 | 含义 |
|---|---|
every <e> | 每次 e 发生时匹配 |
e1 -> e2 | e1 之后发生 e2 (顺序匹配) |
e1 and e2 | 两个事件均发生 (不限顺序) |
e1 or e2 | 其中一个发生 |
not e | e 未发生 |
timer:within(<duration>) | 仅在指定时间内匹配 |
timer:interval(<duration>) | 等待指定时间 |
时间·状态函数
时间函数
| 函数 | 含义 |
|---|---|
current_timestamp | 当前时间 (ms) |
current_date() | 今日日期 (yyyy-MM-dd) |
timestamp(field) | 提取事件的 timestamp |
hour_of_day(ts) | 0~23 时 |
day_of_week(ts) | 1(日)~7(六) |
minute_of_hour(ts) | 0~59 分 |
dayofmonth(ts) | 1~31 日 |
-- 야간 (22시~6시) 발생 알람만
SELECT * FROM Alarm.win:time(1 hour)
WHERE hour_of_day(ts) >= 22 OR hour_of_day(ts) < 6
状态函数
| 函数 | 含义 |
|---|---|
tag('<tag_id>') | 该标签的最新值 (PointMap 上下文) |
prev_status(asset_id) | 资产的上一个状态 |
health(asset_id) | 资产健康状态 (OK/WARN/ERROR/UNKNOWN) |
is_in_shift(asset_id) | 资产是否在班次内 |
算术·字符串函数
| 函数 | 含义 |
|---|---|
abs(x) / round(x, n) / floor(x) / ceil(x) | 数学 |
pow(x, y) | x 的 y 次方 |
sqrt(x) | 平方根 |
length(s) | 字符串长度 |
upper(s) / lower(s) | 大小写转换 |
substring(s, start, len) | 子字符串 |
concat(a, b, ...) | 连接 |
变量与上下文
变量 — 动态阈值
-- 변수 정의 (운영자가 화면에서 조정 가능)
create variable double max_temp = 80;
-- 변수 사용
SELECT * FROM Point.win:time(1 sec)
WHERE tag_id LIKE '%.TEMP' AND value > max_temp
变量值在运行中可以更改，所有 EQL 规则将立即使用新值。
上下文 — 基于时间/条件的激活
Platform 中预定义了常用上下文，可直接使用。
| 上下文 | 激活条件 |
|---|---|
EVERY_1_MINUTES | 每 1 分钟 |
EVERY_5_MINUTES | 每 5 分钟 |
EVERY_10_MINUTES | 每 10 分钟 |
EVERY_30_MINUTES | 每 30 分钟 |
EVERY_1_HOURS | 每 1 小时 |
EVERY_3_HOURS | 每 3 小时 |
EVERY_6_HOURS | 每 6 小时 |
EVERY_12_HOURS | 每 12 小时 |
DAY_WORK_TIME | 工作日 09:00~18:00 |
-- 5분마다 한 번씩 평균 온도 emit (그 사이에는 결과가 안 옴)
context EVERY_5_MINUTES
SELECT avg(value) AS avg_temp
FROM Point.win:time(5 min)
WHERE tag_id = 'MOTOR-001.TEMP'
-- 주간 근무시간에만 평가
context DAY_WORK_TIME
SELECT * FROM AssetAlarm.win:time(1 sec)
WHERE priority = 'ERROR'
12 个实战案例
案例 1 — 实时监控单个标签
SELECT value, ts, quality
FROM Point.win:time(1 sec)
WHERE tag_id = 'MOTOR-001.SPEED'
案例 2 — 立即检测超阈值
SELECT tag_id, value, ts
FROM Point.win:time(1 sec)
WHERE tag_id LIKE '%.TEMP' AND value > 90
案例 3 — 按资产 5 分钟平均 (每 5 分钟发射)
context EVERY_5_MINUTES
SELECT asset_id, avg(value) AS avg_temp, max(value) AS peak
FROM Point.win:time(5 min)
WHERE tag_id LIKE '%.TEMP'
GROUP BY asset_id
案例 4 — 变化量 (对比前值)
SELECT tag_id, value, value - prev(1, value) AS delta
FROM Point.win:length(2)
WHERE tag_id = 'TANK-001.LEVEL'
案例 5 — 仅夜班的 ERROR 报警
SELECT tag_id, priority, band_message, ts
FROM Alarm.win:time(1 sec)
WHERE priority = 'ERROR'
AND hour_of_day(ts) >= 22 OR hour_of_day(ts) < 6
案例 6 — 资产健康状态转移 (OK → WARN)
SELECT asset_id, status, prev_status, ts
FROM AssetHealthStatus.win:time(1 sec)
WHERE prev_status = 'OK' AND status = 'WARN'
案例 7 — 工单 OEE 低下
SELECT order_id, asset_id, oee, availability, performance, quality
FROM OEE.win:time(1 sec)
WHERE oee < 0.6
案例 8 — 检测 30 分钟无响应资产
SELECT * FROM pattern [
every a = AssetData()
-> ( timer:interval(30 min)
and not AssetData(asset_id = a.asset_id) )
]
案例 9 — 启动后立即停止 (运行不稳定)
SELECT * FROM pattern [
every a = AssetEvent(event_type = 'STARTUP')
-> b = AssetEvent(event_type = 'SHUTDOWN', asset_id = a.asset_id)
where timer:within(60 sec)
]
案例 10 — 多标签同时对比 (PointMap)
SELECT
tag('MOTOR-001.TEMP') AS temp,
tag('MOTOR-001.PRESSURE') AS pressure,
tag('MOTOR-001.VIBRATION') AS vibration
FROM PointMap.win:time(1 sec)
案例 11 — 按线单位每小时累计产量
context EVERY_1_HOURS
SELECT asset_id, sum(value) AS hourly_count
FROM AssetAggregation.win:time(1 hour)
WHERE agg_type = 'PRODUCED' AND asset_id LIKE 'LINE-%'
GROUP BY asset_id
案例 12 — TOP 10 报警发生资产
context EVERY_10_MINUTES
SELECT asset_id, count(*) AS alarm_count
FROM AssetAlarm.win:time(10 min)
GROUP BY asset_id
ORDER BY alarm_count DESC
LIMIT 10
性能指南
1. 刻意减小窗口大小
-- ❌ 나쁜 예 — 24시간을 메모리에 유지
SELECT avg(value) FROM Point.win:time(24 hour)
-- ✅ 좋은 예 — 1시간만 메모리에 유지 + context 로 emit 주기 분리
context EVERY_1_HOURS
SELECT avg(value) FROM Point.win:time(1 hour)
2. 将 WHERE 子句靠近触发器
将标签 ID·资产 ID·站点 ID 等静态过滤器放在 WHERE 子句最前面。它们会在随后的 GROUP BY/HAVING 之前被评估。
3. LIKE '%...%' 双面通配符成本较高
-- ❌ 나쁜 예
WHERE tag_id LIKE '%TEMP%'
-- ✅ 좋은 예
WHERE tag_id LIKE '%.TEMP' -- 접미사 매칭만
4. 模式匹配应搭配短窗口
不要遗漏 pattern [ ... timer:within(<n-minutes>) ] 的时间限定符。无限模式会持续占用内存。
5. 调整结果发射频率
通过 context EVERY_N_MINUTES 上下文让 EQL 本身每秒评估一次，但仅每 N 分钟发射一次，以此减轻下游负载。
常见问题
Q. 输入 EQL 查询后结果界面没有任何显示。
A. 请检查：① 是否仅使用了 FROM Point — 需要窗口 (Point.win:time(1 sec))。② WHERE 条件是否太严导致无法匹配。③ 是否点击了右上角的 ▶ 开始按钮。④ 结果面板是否处于 ⏸ 停止状态。
Q. 与 SQL 有什么区别？
A. 最大区别是 窗口 概念。由于 EQL 在无限流上运行，必须指定“哪个范围”。此外，pattern [ ... ] 序列匹配和 context 时间上下文是 SQL 中没有的功能。
Q. win:time(1 sec) 与 win:time_batch(1 sec) 的区别？
A. win:time(1 sec) 是 sliding — 每进入一个事件就更新结果。win:time_batch(1 sec) 是 tumbling — 按 1 秒单位打包，仅在 1 秒到期瞬间发射一次。如果需要精确每 1 秒一次结果，请使用 batch。
Q. pattern [ ... ] 与普通 SELECT ... WHERE ... 的区别？
A. SELECT 观察单个事件 (或窗口) 进行评估，而 pattern 评估事件序列 (A 后 B, A 缺失 B 等)。如果时间顺序·因果关系重要，请使用 pattern。
Q. 注册的变量会立即生效吗？
A. 是的。在界面上更改通过 create variable 定义的变量值后，所有已部署的 EQL 规则从下次评估起将使用新值。
Q. 如何将 EQL 匹配结果发送到报警？ A. 在 CEP → 触发器 中注册 EQL 并开启 “MQ 流” 或 “存储保存” 选项，或者在 报警 → EQL 报警 中注册 EQL，匹配时将自动发布报警。
Q. 能将资产单位规则批量应用于所有资产吗？ A. 在 CEP → 语句 中定义资产单位 EQL 规则，并在模型/本体中将该规则映射到资产类别，即可自动部署到该类别的所有资产。每增加资产时也会自动应用。
Q. 一行 EQL 可以同时处理两个流吗？
A. 可以。如上述 JOIN 节示例，在 FROM 子句中使用逗号列出两个流，并使用 WHERE 编写连接条件即可。
Q. 结果太多了。
A. 可以使用 LIMIT N 仅接收前 N 条，或使用 context EVERY_N_MINUTES 增加发射周期，或使用 win:time_batch 捆绑为批次窗口。使 WHERE 子句更严格地过滤也非常有效。
Q. 在哪里练习编写 EQL？ A. CEP → 查询 界面即为 EQL 实践工具。输入一行并点击 ▶ 运行即可立即获取结果。请从上述 12 个实战案例 开始尝试复制。
相关界面
- CEP — EQL 查询/触发器/语句编写界面
- 报警 → EQL 报警 — 通过 EQL 发布任意条件报警
- 流 —
flow_script_filter/flow_script_transform/flow_switch的 EQL 模式 - 标签样式 — 人工智能页签 — 预测·异常检测规则的 EQL 条件
- 域 ID 综合搜索 — 快速查询 EQL 使用的标签/资产 ID (Ctrl+K)