基于时间的siddhi数据聚合

Question

我试图根据时间窗口聚合传感器数据，并在它达到30秒窗口（卷起）后将其写入Cassandra。

例如，名为“temp”的传感器发送3个读数30秒。我喜欢在最后30秒内获得此传感器的平均值，并在窗口完成时将avg值写入Cassandra。

这是我的代码

BasicConfigurator.configure();


        // Create Siddhi Application
        String siddhiApp = "define stream SensorEventStream (sensorid string, value double); " +
                " " +
                "@info(name = 'query1') " +
                "from SensorEventStream#window.time(30 sec)  " +
                "select sensorid, avg(value) as value " +
                "group by sensorid " +
                "insert into AggregateSensorEventStream ;";

        // Creating Siddhi Manager
        SiddhiManager siddhiManager = new SiddhiManager();

        //Generating runtime
        SiddhiAppRuntime siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(siddhiApp);

        //Adding callback to retrieve output events from query
        siddhiAppRuntime.addCallback("AggregateSensorEventStream", new StreamCallback() {


            @Override
            public void receive(org.wso2.siddhi.core.event.Event[] events) {
                 EventPrinter.print(events);
            }
        });

        //Retrieving input handler to push events into Siddhi
        InputHandler inputHandler = siddhiAppRuntime.getInputHandler("SensorEventStream");

        //Starting event processing
        siddhiAppRuntime.start();

        //Sending events to Siddhi
        inputHandler.send(new Object[]{"Temp", 26d});
        Thread.sleep(1000);
        inputHandler.send(new Object[]{"Temp", 25d});
        Thread.sleep(1000);
        inputHandler.send(new Object[]{"Temp", 24d});
        Thread.sleep(60000);
        inputHandler.send(new Object[]{"Temp", 23d});

        //Shutting down the runtime
        siddhiAppRuntime.shutdown();

        //Shutting down Siddhi
        siddhiManager.shutdown();

输出是这样的

0 [main] INFO org.wso2.siddhi.core.util.EventPrinter  - [Event{timestamp=1552281656960, data=[Temp, 26.0], isExpired=false}]
1002 [main] INFO org.wso2.siddhi.core.util.EventPrinter  - [Event{timestamp=1552281657971, data=[Temp, 25.5], isExpired=false}]
2003 [main] INFO org.wso2.siddhi.core.util.EventPrinter  - [Event{timestamp=1552281658972, data=[Temp, 25.0], isExpired=false}]
62004 [main] INFO org.wso2.siddhi.core.util.EventPrinter  - [Event{timestamp=1552281718972, data=[Temp, 23.0], isExpired=false}]

从这个演示代码我看到它立即发送3个事件的第一个临时平均值，在30秒窗口后，它什么也没做。然后打印23。

如何在30秒后窗口卷起时收到通知？我认为这就是接收功能所做的事情。

我不确定我是否误解了这里的功能。这有可能与siddi一起吗？

Answer 1

这是预期的行为，窗口是一个滑动窗口。这里，当第一个事件到来时，第一个秒，窗口仅保持第一个事件，因此平均值为26.然后当第二个事件到达时，窗口同时具有26d和25d，然后是25.5的平均值。同样，第3秒平均25d。然后在31,32和33秒这些事件将从窗口到期。因此，当您的第4个事件到来时（第63秒），窗口中只有最新事件，因此平均值将是值本身。该窗口根据事件到达之前的最后30秒内收到的事件，在事件到达时立即计算平均值。

从你的问题来看，你似乎想要timeBatch窗口。此处，仅在批次结束时计算平均值。例如，在这种情况下，第30,60,90秒等等。有关样本，请参阅timeBatch doc。