Siddhi cep

define stream TemperatureStream
        (sensorId string, temperature double);
 
@info(name = 'Overall-analysis')
from TemperatureStream#window.timeBatch(1 min)
select avg(temperature) as avgTemperature,
       max(temperature) as maxTemperature,
       count() as numberOfEvents
insert into OverallTemperatureStream;
 
@info(name = 'SensorId-analysis')
from TemperatureStream#window.timeBatch(30 sec, 0)
select sensorId,
       avg(temperature) as avgTemperature,
       min(temperature) as maxTemperature
group by sensorId
having avgTemperature > 20.0
insert into SensorIdTemperatureStream;

define stream InputStream(jsonString string);
 
from InputStream
select json:toObject(jsonString) as jsonObj
insert into PersonalDetails;
 
from PersonalDetails
select jsonObj,
    json:getString(jsonObj,'$.name') as name,
    json:isExists(jsonObj, '$.salary') as isSalaryAvailable,
    json:toString(jsonObj) as jsonString
insert into OutputStream;
 
from OutputStream[isSalaryAvailable == false]
select
    json:setElement(jsonObj, '$', 0f, 'salary') as jsonObj
insert into PreprocessedStream;

TemperatureStream  (
        sensorId  string ,  temperature  double );
 
define table TemperatureLogTable (
        sensorId string, roomNo string, temperature double);
 
@store(type="rdbms",
       jdbc.url="jdbc:mysql://localhost:3306/sid",
       username="root", password="root",
       jdbc.driver.name="com.mysql.jdbc.Driver")
define table SensorIdInfoTable (
        sensorId string, roomNo string);
 
@info(name = 'Join-query')
from TemperatureStream as t join SensorIdInfoTable as s
     on t.sensorId == s.sensorId
select t.sensorId as sensorId, s.roomNo as roomNo
       t.temperature as temperature
insert into TemperatureLogTable;