WSO2: Обновление выходного потока в запросе Сиддхи

Question

У меня есть запрос сиддхи, чтобы получить общее количество событий за одну минуту, используя окно времени. Используя выходной поток, я обновляю гистограмму с постоянными приближающимися значениями, отметками времени (дата и время, минута) на оси х и событиями, отсчитываемыми по оси Y.

Но иногда бывает, что либо количество событий за одну минуту занимает слишком много времени для передачи, поэтому запрос не дает правильных результатов.

Например, если я получаю итоговые 60 событий, и этот запрос сначала дает мне счет 40, который отображается на гистограмме, но затем через минуту он меняет свое значение на 20, что является правильным в соответствии с логикой, но я обеспокоен тем, есть способ, которым я мог бы обновить поток, а также гистограмму для любых предыдущих временных меток (в этом случае 40 + 20) и вставить в нее новые значения для следующих предстоящих временных меток.

Я видел, что функция обновления используется с таблицами, а не потоком, не так ли? А также я хочу, чтобы 2 outputStreams заполняли две разные гистограммы из одного входного потока. Итак, верно ли правильное предложение для этой цели?

Запрос является:

/* 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

Но иногда есть когда число событий в одну минуту слишком долго, чтобы быть передана и, следовательно, запрос не дает правильные результаты.

В приведенной выше ситуации вы, вероятно, должны использовать окно externalTimeBatch, а не окно timeBatch. Затем он будет обрабатывать пакет событий относительно временной метки события вместо фактического времени.

Я видел, что функция обновления используется с таблицами, а не потоком, не так ли?

Да. только таблицы событий имеют функцию удаления/обновления. Потоки событий не имеют такой функциональности.

А также я хочу, чтобы 2 выходных потока заполняли две разные гистограммы из одного входного потока. Итак, верно ли правильное предложение для этой цели?

Да. Поскольку вы экспортируете FTPOutStream и FTPIPOutStream, вы можете использовать их для заполнения двух разных гистограмм.