Я хочу непрерывно вычислять% протоколов, появляющихся в сетевом трафике, чтобы эти% постоянно обновлялись новыми событиями. Создается круговая диаграмма, которая обновляется процентами. Поскольку для расчета мне нужны как новые, так и предыдущие данные, я решил использовать таблицу в памяти, чтобы хранить события более длительное время (скажем, сутки).
Поскольку таблицы событий можно использовать только при соединении с потоками событий, я выбрал внешнее соединение, чтобы также получить старые значения. Поскольку меня интересуют только протоколы и их проценты, мне нужно всего два столбца, но я не могу применить агрегатную функцию во внешнем соединении. Запрос, который я до сих пор сгенерировал:
@Import('MAINInStream:1.0.0')
define stream MAINInStream (ts string, uid string, id_orig_h string, id_orig_p int, id_resp_h string, id_resp_p int, proto string, service string, duration double, orig_bytes long, resp_bytes long, conn_state string, local_orig bool, local_resp bool, missed_bytes long, history string, orig_pkts long, orig_ip_bytes long, resp_pkts long, resp_ip_bytes long, tunnel_parents string, sensorname string);
@Export('ProtocolStream:1.0.0')
define stream ProtocolStream (protocol string, count int);
define table mem_conn_table (timestamp long, id_orig_h string, id_orig_p int, id_resp_h string, id_resp_p int, proto string);
from MAINInStream
select time:timestampInMilliseconds(time:dateAdd(str:replaceAll(ts,'T',' '), 5, 'hour',"yyyy-MM-dd HH:mm:ss"),'yyyy-MM-dd HH:mm') as timestamp, id_orig_h, id_orig_p, id_resp_h, id_resp_p, proto
insert into intermediateStream;
from MAINInStream
select time:timestampInMilliseconds(time:dateAdd(str:replaceAll(ts,'T',' '), 5, 'hour',"yyyy-MM-dd HH:mm:ss"),'yyyy-MM-dd HH:mm') as timestamp, id_orig_h, id_orig_p, id_resp_h, id_resp_p, proto
group by id_resp_p
insert into mem_conn_table;
from intermediateStream#window.externalTimeBatch(timestamp,1min, timestamp, 1min) as i right outer join mem_conn_table[time:dateDiff(time:currentTimestamp(),cast(timestamp,"string"), "yyyy-MM-dd HH:mm:ss", "yyyy-MM-dd HH:mm:ss") == 0] as mc
on i.timestamp == mc.timestamp
SELECT (ifThenElse(mc.id_resp_p == 21,'FTP', ifThenElse(mc.id_resp_p == 22,'SSH', ifThenElse(mc.id_resp_p == 25,'SMTP', ifThenElse(mc.id_resp_p == 445,'SMB','MYSQL'))))) as protocol , cast(count(mc.id_resp_p),'int') as count
insert into ProtocolStream;
Я группирую окно с одной внешней минутой, а затем получаю протоколы и их количество, но это не дает мне никакого результата.
Какие-либо предложения?
#window.externalTimeBatch(timestamp,1min, timestamp, 1min)
в последнем запросе? - person Grainier   schedule 29.08.2016