WSO2: Aktualisieren eines Ausgabestreams in Siddhi-Abfrage

Question

Ich habe eine Siddhi-Abfrage, um die Anzahl der Gesamtereignisse in einer Minute mithilfe eines Timebatch-Fensters abzurufen. Unter Verwendung eines Ausgabestroms aktualisiere ich ein Balkendiagramm mit konstanten kommenden Werten, Zeitstempeln (Datum und Zeit, bis Minute) auf der X-Achse und Ereignissen auf der Y-Achse.

Es gibt jedoch manchmal, wenn entweder die Anzahl der Ereignisse in einer Minute zu lange dauert, um übertragen zu werden, und die Abfrage daher keine korrekten Ergebnisse liefert.

Zum Beispiel, wenn ich insgesamt 60 Ereignisse bekomme und diese Abfrage gibt mir zuerst die Anzahl von 40, die in Balkendiagramm angezeigt wird, aber nach einer Minute ändert es seinen Wert auf 20, die nach der Logik richtig ist, aber ich bin besorgt ob Es gibt eine Möglichkeit, den Stream und das Balkendiagramm für alle vorherigen Zeitstempel (in diesem Fall 40 + 20) zu aktualisieren und neue Werte für die nächsten kommenden Zeitstempel einzufügen.

Ich habe gesehen, Update-Funktion wird mit Tabellen nicht Stream verwendet, ist es so? Und ich möchte auch, dass 2 outputStreams zwei verschiedene Balkendiagramme aus dem gleichen Eingabestream bevölkern. Also ist unten die Abfrage für diesen Zweck korrekt?

Abfrage ist:

/* Enter a unique ExecutionPlan */ 
@Plan:name('FTPExecutionPlan') 

/* Enter a unique description for ExecutionPlan */ 
-- @Plan:description('ExecutionPlan') 

/* define streams/tables and write queries here ... */ 

@Import('FTPInStream:1.0.0') 
define stream FTPInStream (ts string, uid string, id_orig_h string, id_orig_p int, id_resp_h string, id_resp_p int, user string, password string,command string, arg string, mime_type string, file_size string, reply_code int, reply_msg string); 

@Export('FTPIPOutStream:1.0.0') 
define stream FTPIPOutStream (ip_address string, ftp_requests int); 

@Export('FTPOutStream:1.0.0') 
define stream FTPOutStream (ts string, ftp_requests int); 

from FTPInStream 
select time:dateFormat(str:replaceAll(ts,'T',' '),'yyyy-MM-dd HH:mm', 'yyyy-MM-dd HH:mm:ss') as ts, uid, id_orig_h, id_orig_p, id_resp_h, id_resp_p 
insert into intermediateStream; 

from intermediateStream#window.timeBatch(1 min) 
select ts, cast(count(ts), 'int') as ftp_requests 
group by ts 
insert into FTPOutStream; 

from intermediateStream#window.timeBatch(1 min) 
select id_orig_h as ip_address, cast(count(id_orig_h), 'int') as ftp_requests 
group by id_orig_h 
insert into FTPIPOutStream; 

Answer 1

Aber es gibt manchmal, wenn entweder die Anzahl der Ereignisse in einer Minute zu lange dauern übertragen zu werden und somit nicht Abfrage korrekte Ergebnisse nicht geben.

In obiger Situation, sollten Sie wahrscheinlich externalTimeBatch Fenster verwenden, statt timeBatch Fenster. Dann verarbeitet es Ereignisbatch relativ zum Zeitstempel des Ereignisses anstelle der tatsächlichen Zeit.

Ich habe gesehen, Update-Funktion wird mit Tabellen nicht Strom verwendet, ist es so?

Ja. Nur Ereignistabellen verfügen über Lösch-/Aktualisierungsfunktionen. Ereignisströme haben diese Funktionalität nicht.

Und ich möchte auch 2 outputStreams bevölkern zwei verschiedene Balkendiagramme aus dem gleichen Eingabestrom. Also ist unten die Abfrage für diesen Zweck korrekt?

Ja. Da Sie sowohl FTPOutStream als auch FTPIPOutStream exportieren, können Sie zwei unterschiedliche Balkendiagramme füllen.