使用WSO2更新MySQL數據庫CEP

Question

有什麼方法可以在不更新流中的空值的情況下更新MySQL Db。如果我的輸入數據流包含一些空值，則當前使用「data_empty」值表示空值。那時CEP用該值更新數據庫（「data_empty」）。我的目標是不更新空值更新其餘的東西。是否有可能使用siddhi和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'）更新到數據庫是很困難的。然而在Siddhi，有一個叫做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;