跳到主要内容

EQL — 事件查询语言帮助

目录​

概述​

主要使用位置 左侧菜单 Automation > EQL 查询(/query/index) 直接运行，同样的语法也用于高级 EQL 报警 · 触发器·声明 · 流 的条件表达式中。

EQL (Event Query Language) 是 PlantPulse 平台自创的查询语言，让您能像使用 SQL 那样对实时流数据进行查询、聚合和模式匹配。普通 SQL 作用于静态表，而 EQL 在持续流入的事件流上运行，每一刻都会产生结果。

运营者可以用一条 EQL 语句实现以下功能：

  • 实时监控 — 标签值超过阈值的瞬间立即输出结果
  • 资产级聚合 — 每5分钟自动计算平均值/最大值/最小值
  • 模式匹配 — "事件 A 后30秒内没有事件 B 则报警"
  • 时序联接 — 两个标签的时间对齐比较

EQL 使用几乎与 SQL 相同的语法，但额外引入了窗口(Window) 概念，用来指定"最近 N 秒 / N 条"这样的时间或长度范围。

在哪里使用 EQL​

平台内 EQL 的使用位置如下。

EQL 查询 · 触发器 · 声明在左侧菜单 Automation 部分并排显示。 (没有名为 CEP 的菜单 — 旧文档把这三项列为 CEP 下的屏幕。)

菜单 · 页面内部 URL用途
Automation > EQL 查询 — 文档/query/index运营者即时运行 EQL 并查看结果（探索·调试）
Automation > 触发器 — 文档/trigger/index自动将 EQL 匹配结果发布到 MQ 主题或存储
Automation > 声明 — 文档/statement/index按资产注册的规则（状态/聚合/事件/命令）
报警 → EQL 报警—用 EQL 规则发布任意条件报警
流 → 脚本—flow_script_filter / flow_script_transform / flow_switch 的 EQL 表达式模式
标签表单 — AI 标签页—预测·异常检测规则的输入条件

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赋予别名
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 子句可以使用的流（事件源）如下。所有流都由平台自动生成，无需额外注册即可立即使用。

使用此列表之外的名称在 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(键值对), ts
AssetEvent资产事件发生（启动·停止·维护等）asset_id, event_type, details, ts
AssetAlarm资产报警发生asset_id, alarm_band, priority, message
AssetCommand资产下发命令时asset_id, command, args
AssetAggregation1分/5分/1小时资产聚合结果asset_id, window, agg_type, values
AssetContext资产上下文(元数据)变更asset_id, context_key, before, after
AssetHealthStatus · AssetConnectionStatus 不是流

旧文档中这两项被列为资产域流，但它们不是注册的事件类型。 在 FROM AssetHealthStatus... 中使用它们会导致规则部署失败。

健康·连接状态评估不是流，而是以定期执行作业方式运行 (AssetHealthStatusDeployer / AssetConnectionStatusDeployer)。其结果可以在页面和 API 中查看。如果想在 EQL 中捕捉状态变化，应使用原始流 AssetEvent · AssetAlarm 或 Point 的未接收，并用not 模式检测。

生产域 (3种)​

流触发时刻主要字段
Calendar班次启动/结束/变更shift_id, calendar_id, event, start_time, end_time
WorkOrder工作单生命周期转移order_id, status, prev_status, asset_id, product_id
OEEOEE 评估结果更新order_id, asset_id, oee, availability, performance, quality

系统域 (2种)​

流触发时刻主要字段
Log系统诊断消息module, level, code, summary, ts
OSPerformance服务器操作系统资源指标（周期）CPU·内存·磁盘等
不存在 Status 流

事件类型 Status 未在注册列表中。如果想用 EQL 捕捉 OPC·边缘的连接状态变化， 这个流无法使用 — 应该用not 模式检测 Point 的未接收。

内部流 (3种)​

已注册但不用于规则编写。用于诊断·健康检查。

流用途
DateTime定期触发的时间事件（基于时间的规则触发）
Ping引擎生存确认
Test规则验证用

所有流必须先与窗口结合使用才能用（如 Point.win:time(1 sec)）。仅使用 FROM Point 而不加窗口， 结果会立即消失而看不见。

窗口​

窗口是"从流中截取的时间或长度段"。EQL 支持两种窗口类型。

时间窗口 (Time Window)​

语法含义
Point.win:time(1 sec)最近1秒内收到的所有事件（滑动）
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)最近1000条

批处理窗口 (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比较
BETWEENvalue BETWEEN 70 AND 90范围
INtag_id IN ('A','B','C')列表包含
LIKEtag_id LIKE 'MOTOR-%'模式匹配（% 通配符）
IS NULL / IS NOT NULLquality IS NOT NULLnull 判断
AND OR NOTvalue > 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 ee 未发生
timer:within(<duration>)仅在指定时间内匹配
timer:interval(<duration>)等待指定时长

时间·状态函数​

时间函数​

函数含义
current_timestamp当前时刻（毫秒）
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 规则立即应用新值。

上下文 — 基于时间/条件的激活​

平台中预定义了常用上下文，您可以直接使用。

上下文激活条件
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. EQL 与 SQL 有什么不同？ A. 最大的差异是窗口概念。EQL 在无限流上运行，所以必须指定"哪个范围"。另外 pattern [ ... ] 序列匹配和 context 时间上下文是 SQL 中没有的功能。

Q. win:time(1 sec) 和 win:time_batch(1 sec) 的区别是什么？ A. win:time(1 sec) 是滑动 — 每有事件进入时结果就更新。win:time_batch(1 sec) 是翻滚 — 按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选复制开始。

相关页面​