Обновление базы данных MySQL с использованием WSO2 CEP

Question

Есть ли способ обновить MySQL Db без обновления пустых значений в потоке. Если мой поток входных данных содержит несколько пустых значений, в настоящее время это пустое значение указывает с использованием значения «data_empty». В это время CEP обновляет БД этим значением («data_empty»). Моя цель заключается в том, что обновление этого пустого значения не обновляется. Можно ли использовать сиддхи и WSO2 CEP.

@Plan:name('DBUpdateExecutionPlan') 
@Import('testStream:1.0.0') 
define stream input (id string, param1 string, param2 string); 

@Export('testOutStream:1.0.0') 
define stream output (id string, param1 string, param2 string); 

@from(eventtable = 'rdbms' , datasource.name = 'MYSQL' , table.name = 'cep') 
define table cepTable (id string, param1 string, param2 string) ; 

from input#window.time(0 sec) 
select * 
update cepTable on id == cepTable.id; 

Answer 1

Это своего рода трудно обновить только непустое (не «data_empty» в вашем сценарии) значения в базе данных. Однако в Сиддхи есть функция, называемая ifThenElse(condition, value if true, value if false), которая может использоваться в вашем сценарии. Обратитесь к приведенному ниже образцу плана выполнения, чтобы получить представление об использовании ifThenElse() и обновлении таблицы (аналогично вашей учетной записи).

@Plan:name('IfThenElseExecutionPlan') 

@Import('inputStream:1.0.0') 
define stream dataIn (roomId int, roomType string, roomTemp float); 

@Export('outputStream:1.0.0') 
define stream dataOut (roomId int, roomType string, roomTemp float); 

@From(eventtable='rdbms', datasource.name='cepdatabase', table.name='roomTable') 
define table roomTable (roomId int, roomType string, roomTemp float); 

from dataIn[not((roomTable.roomId == roomId) in roomTable)] 
insert into updateStream; 

from dataIn join roomTable 
on roomTable.roomId == dataIn.roomId 
select dataIn.roomId as roomId, 
     ifThenElse(dataIn.roomType=='data_empty', roomTable.roomType, dataIn.roomType) as roomType, 
     ifThenElse(dataIn.roomTemp==0.0f, roomTable.roomTemp, dataIn.roomTemp) as roomTemp 
insert into updateStream; 

from updateStream 
insert overwrite roomTable 
on roomTable.roomId == roomId;