Поток - объединить данные о похожих событиях в одно событие - PullRequest
Новая
       9

Поток - объединить данные о похожих событиях в одно событие

0 голосов
/ 22 октября 2018

Я хотел бы объединить входящие события в одно событие на основе одного из полей.

Входные события: 

{
  ID: '123',
  eventType: 'a',
  eventCode: 1
},
{
  ID: '123',
  eventType: 'b',
  eventCode: 2
},
{
  ID: '123',
  eventType: 'c',
  eventCode: 3
}

Ожидаемый результат: 

{
  ID: '123',
  events: [{
    eventType: 'a',
    eventCode: 1
  },
  {
    eventType: 'b',
    eventCode: 2
  },
  {
    eventType: 'c',
    eventCode: 3
  }]
}

Я группирую события на основе окна 4. Итак, мне нужно обработать 4 события, объединить их и передать на следующий шаг.

Вариант использования : я быхотел бы использовать сгенерированный вывод для хранения в MongoDB ИЛИ передать его во внешнюю службу.

Возможно ли это с помощью Siddhi?

ПРИМЕЧАНИЕ : я вижу, что a аналогичный вопрос уже задавался, но ответ пришел 5 лет назад, и с тех пор Сиддхи прошел долгий путь.

1 Ответ

0 голосов
/ 25 октября 2018

Вы можете использовать приведенные ниже приложения сиддхи, чтобы удовлетворить ваши требования.Я использовал расширение строки , чтобы сделать это.Но учтите, что сгенерированный вывод - именно тот, который вы запрашивали.Если вы хотите получить правильный вывод JSON, вам также может понадобиться выполнение json extention .Следуйте инструкциям, чтобы узнать подробнее об использовании расширений. 

@App:name("testJsonConcat")
@App:description("Description of the plan")

-- Please refer to https://docs.wso2.com/display/SP400/Quick+Start+Guide on getting started with SP editor. 

define stream inputStream(id string, eventType string, eventCode int);

partition with (id of inputStream)
begin
from inputStream
select id, str:concat("{eventType: '", eventType, "' , eventCode :",eventCode,"}") as jsonString
insert into #formattedStream;

from #formattedStream#window.lengthBatch(4)
select str:concat("{ ID : '", id, "',events: [", str:groupConcat(jsonString),"]}") as result
insert into concatStream;
end;

from concatStream#log()
select *
insert into temp;
...