wso2 cep Siddhiql

Question

私は、複数のセンサーが測定イベントをストリームに送信しています。イベントは{パラメータ、値、タイムスタンプ}で構成されます。私は数日間の時間枠でこれらの値を観察し、傾向を確認し、これらの異なるセンサーによって監視されている機器に関する診断を行いたいと思います。

1. ストリームをパラメータで分割します。受信される各パラメータの

 
    from inputStream[parameter='A'] 
    select * 
    insert into Astream; 

など。

2. timewindowの場合、60秒間、線形回帰を計算して変更を検出します。

 
    from Astream#timeseries:lengthTimeRegress(60000, value, timestamp) 
    select beta1 * 100 as AChange 
    insert into AChangeStream; 

この私が各メトリックストリームのために行います。 3.各ストリームのトレンドを取得したら、各ストリームの変更された値を収集し、条件を満たすかどうかを確認します。

 
    from every e1=AChangeStream[e1.AChangeStream > 0.5], e2=BChangeStream[e2.BChangeStream 0.15] 
    select 'condition 1 alarm' as message 
    insert into alertStream; 

上記のsiddhi qlは、時間ウィンドウ内の6つのパラメータの変化を検出しますか？

Answer 1

あなたが提供したクエリの要点は、あなたが逃した小さな事を除いて正しいです。 6つのパラメータを言うとき、私はあなたがパラメータA、B、C、D、E、Fに類似したものを持っていると信じています。最終的に、与えられた条件[1]に一致する一連のイベントを見つけたいと思われます。

パラメータAとBの2つのみを考慮すると、以下のようにSiddhi言語でクエリを書くことができます。

 

@Import('input:1.0.0') 
define stream inputStream (parameter string, value double, timestamp long);  


from inputStream[parameter=='A'] 
select * 
insert into Astream;  

from inputStream[parameter=='B'] 
select * 
insert into Bstream;  

from Astream#timeseries:lengthTimeRegress(60000, 10000, value, timestamp) 
select beta1*100 as AChange 
insert into AChangeStream;  

from Bstream#timeseries:lengthTimeRegress(60000, 10000, value, timestamp) 
select beta1*100 as BChange 
insert into BChangeStream;  

from every e1=AChangeStream[e1.AChange > 0.5], e2=BChangeStream[e2.BChange > 0.15] 
select 'condition 1 alarm' as message 
insert into alertStream; 

以下に注意してください。

1. lengthTimeRegress関数では、[2]で指定されているように、4つの必須パラメータを指定する必要があります。あなたが書いたクエリのバッチサイズが不足しています。回帰計算に使用されるイベントの最大数は、バッチサイズによって指定されます。
2. シーケンス条件では、パラメータを使用する必要があります。ストリーム名ではありません。あなたが誤っ> e1.AChangeStreamとして書かれている何0.5> 0.5

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

e1.AChangeとして変更する必要があります