wso2 cep Siddhiql

Question

我有多个传感器将测量事件发送到流。一个事件由{参数、值、时间戳}组成。我想在几天的时间窗口内观察这些值并检查趋势并对这些不同传感器监控的设备进行诊断。

1. 按参数划分流。

来自 inputStream[参数='A'] 选择 * 插入Astream；

对于接收到的每个参数，依此类推。

1. 对于时间窗口，比如 60 秒，计算线性回归以找出变化。

来自 Astream#timeseries:lengthTimeRegress(60000, value, timestamp) 选择 beta1 * 100 作为 AChange 插入AChangeStream；

我对每个指标流都这样做。 3. 获得每个流的趋势后，我会收集每个流的更改值并检查它们是否满足条件。

从每个 e1=AChangeStream[e1.AChangeStream > 0.5], e2=BChangeStream[e2.BChangeStream 0.15] 选择“条件 1 警报”作为消息 插入警报流；

上面的siddhi ql会检测到时间窗内6个参数的变化吗？

Answer 1

您提供的查询的要点是正确的，除了您遗漏了一些小事。当你说 6 个参数时，我相信你有类似于参数 A、B、C、D、E 和 F 的东西。最终你似乎想要找到与给定条件匹配的事件序列 [1]。

仅考虑 2 个参数 A 和 B，您可以用 Siddhi 语言编写查询，如下所示，以满足您的要求。

@Import('输入：1.0.0') 定义流inputStream（参数字符串，值double，timestamp long）； 来自 inputStream[参数=='A'] 选择 * 插入Astream； 来自 inputStream[参数=='B'] 选择 * 插入Bstream； 来自 Astream#timeseries:lengthTimeRegress(60000, 10000, value, timestamp) 选择 beta1*100 作为 AChange 插入AChangeStream； 来自 Bstream#timeseries:lengthTimeRegress(60000, 10000, value, timestamp) 选择 beta1*100 作为 BChange 插入 BChangeStream； 从每个 e1=AChangeStream[e1.AChange > 0.5], e2=BChangeStream[e2.BChange > 0.15] 选择“条件 1 警报”作为消息 插入警报流；

请注意以下事项。

1. 在lengthTimeRegress 函数中，您需要提供4 个强制参数，如[2] 中所述。您在编写的查询中错过了批量大小。用于回归计算的最大事件数由批量大小指定。
2. 在序列条件中，需要使用参数。不是流名称。您错误地写为 e1.AChangeStream > 0.5 的内容必须更改为 e1.AChange > 0.5

[1]https://docs.wso2.com/display/CEP420/SiddhiQL+Guide+3.1#SiddhiQLGuide3.1-Sequence
[2]https://docs.wso2.com/display/SIDDHIEXTENSIONS/Regression