Сравнение длинных пакетов данных друг с другом в Сиддхи - PullRequest
Новая
       60

Сравнение длинных пакетов данных друг с другом в Сиддхи

0 голосов
/ 15 октября 2019

Я определил входной поток ниже. Строка Datetime похожа на 2010-09-01 06:59:00.000, результат в два раза больше, чем 157,382, а UnixDateTime имеет тип long типа 1283324340111. 

define stream HStream(ID int, DateTime String, Result double, UnixDateTime long);

Я хочу создать пакеты длины для 100 отображаемых событийсреднее для столбца result, и я хочу сравнить эти партии друг с другом. Я хочу сделать это скользящее сравнение для следующих 5 партий (каждый из которых включает в себя 100 событий). Поэтому я хочу сравнить первую партию (событие 0-100) со второй партией (101-200), до шестой (501-600). И я хочу сравнить вторую партию до 7-й партии. Что я хочу достичь с помощью сравнения, так это то, что когда 4 или более (из 5) партий имеют средний результат партии, который все больше или меньше 1 (по сравнению со средним результатом из исходной партии), тогда я хочуЗаписать информацию об оригинальной партии.

Мой код указан ниже. Проблема, я не знаю точный синтаксис. Я посмотрел учебники и документацию по WSO2 и Siddhi, но не могу решить проблему. 

@info(name = 'MovingAverageQuery')
from every e1=HStream, e2=HStream[e1.avg(Result) <= avg(Result))+, e2=HStream[e2[last].avg(Result) <= avg(Result)]
select ID, DateTime, Result, 
avg(Result), UnixDateTime
output last every 100 events
insert into OutputStream;

@sink(type='log', prefix='LOGGER')
define stream OutputStream(Nr ID, DateTime String, Result double, Avg double, UnixDateTime long);

1 Ответ

0 голосов
/ 16 октября 2019

Для требования необходимо использовать два запроса: один для вычисления средних значений (Average100Query), а другой - для сравнения средних значений (IdentifyIncreaseingTrend). 

@App:name("AverageSequence")
@App:description("Identify the average increase trend")

define stream HStream(ID int, DateTime String, Result double, UnixDateTime long);

@sink(type='log', prefix='LOGGER')
define stream OutputStream(ID int, DateTime String, avgResult double, UnixDateTime long);

@info(name = 'Average100Query')
from HStream#window.lengthBatch(100)
select ID, DateTime, avg(Result) as avgResult, UnixDateTime 
insert into AverageStream;

@info(name='IdentifyIncreaseingTrend')
from every e1=AverageStream, e2=AverageStream[e2.avgResult >= (e1.avgResult + 1)],  e3=AverageStream[e3.avgResult >= (e2.avgResult + 1)],  e4=AverageStream[e4.avgResult >= (e3.avgResult + 1)], e5=AverageStream[e5.avgResult >= (e4.avgResult + 1)]
select e1.ID, e1.DateTime, e1.avgResult, e1.UnixDateTime 
insert into OutputStream;

Некоторые синтаксические проблемы, которые я заметил, связаны с выполнением вычисления, например, сумма (результат). Вам необходимо использовать ключевое слово as, чтобы присвоить этому атрибуту значение sum(result) as totalResult. В последовательностях вы не можете использовать функцию усреднения, поскольку это необходимо сделать для нескольких событий, но вы можете использовать переименованный атрибут totalResult.

...