Siddhi延迟查询

Question

我很难理解这个查询：

from heartbeats#window.time(1 hour) insert expired events into delayedStream;

from every e = heartbeats -> e2 = heartbeats[deviceId == e.deviceId]
  or expired = delayedStream[deviceId == e.deviceId]
within 1 hour 10 minutes 
select e.deviceId, e2.deviceId as id2, expired.deviceId as id3
insert into tmpStream;

第一个查询将所有事件延迟 1 小时。
第二个查询筛选 1 小时前发生的所有事件，并且没有找到更新的事件。

这有效，但我不明白这部分：

from every e = heartbeats -> e2 = heartbeats[deviceId == e.deviceId] or expired = delayedStream[deviceId == e.deviceId]

查询的第二部分（或 expired = ...）检查具有给定 deviceId 的事件是否在 delayedStream 上。第一部分的目的是什么？它是如何结合在一起的，即此查询查找超过 1 小时未发送数据的设备？

Answer 1

如果您想检查传感器在过去 1 小时内是否未发送读数，我认为上述查询不会准确。我将 windows 调整为 1 分钟并发送了 2 个事件，

[2019-07-19 16:48:23,774] heartbeats : Event{timestamp=1563535103772, data=[1], isExpired=false}
[2019-07-19 16:48:24,696] tmpStream : Event{timestamp=1563535104694, data=[1, 1, null], isExpired=false}
[2019-07-19 16:48:24,697] heartbeats : Event{timestamp=1563535104694, data=[1], isExpired=false}
[2019-07-19 16:49:23,774] tmpStream : Event{timestamp=1563535163772, data=[1, null, 1], isExpired=false}

假设事件在 10 点和 10.15 点到达，tmpStream 的输出将在 10.15 点（第一部分）和 11 点（由于流延迟）。第二个匹配项不正确，因为它必须根据用例匹配 11.15。

但是，如果您想改进查询，您可以为您的用例使用 Siddhi 检测 non-occurance 模式功能，https://siddhi.io/en/v5.0/docs/query-guide/#detecting-non-occurring-events，它会更简单

Siddhi延迟查询

Siddhi delayed query

complex-event-processing

siddhi

wso2sp