Use Rule pipeline to develop a system that detects speed changes.

[Richapagare]

This system should be capable of capturing and storing information from 10 events before a speed change occurs, the event where the speed change happens, and 10 events following the speed change. The system should process this data using ekuiper rules.
sample input data:

speed =65, speed_diff=0, time=1
speed =65, speed_diff=0, time=2
speed =65, speed_diff=0, time=3
speed =65, speed_diff=0, time=4
speed =65, speed_diff=0, time=5
speed =65, speed_diff=0, time=6
speed =65, speed_diff=0, time=7
speed =65, speed_diff=0, time=8
speed =65, speed_diff=0, time=9
speed =65, speed_diff=0, time=10
speed =66, speed_diff=1, time=11
speed =66, speed_diff=0, time=12
speed =66, speed_diff=0, time=13
speed =66, speed_diff=0, time=14
speed =66, speed_diff=0, time=15
speed =66, speed_diff=0, time=16
speed =66, speed_diff=0, time=17
speed =66, speed_diff=0, time=18
speed =66, speed_diff=0, time=19
speed =66, speed_diff=0, time=20
speed =66, speed_diff=0, time=21
speed =66, speed_diff=0, time=22
speed =66, speed_diff=0, time=23
speed =66, speed_diff=0, time=24
speed =66, speed_diff=0, time=25
speed =66, speed_diff=0, time=26....

expected output:
speed =65, speed_diff=0, time=1
speed =65, speed_diff=0, time=2
speed =65, speed_diff=0, time=3
speed =65, speed_diff=0, time=4
speed =65, speed_diff=0, time=5
speed =65, speed_diff=0, time=6
speed =65, speed_diff=0, time=7
speed =65, speed_diff=0, time=8
speed =65, speed_diff=0, time=9
speed =65, speed_diff=0, time=10
speed =66, speed_diff=1, time=11
speed =66, speed_diff=0, time=12
speed =66, speed_diff=0, time=13
speed =66, speed_diff=0, time=14
speed =66, speed_diff=0, time=15
speed =66, speed_diff=0, time=16
speed =66, speed_diff=0, time=17
speed =66, speed_diff=0, time=18
speed =66, speed_diff=0, time=19
speed =66, speed_diff=0, time=20
speed =66, speed_diff=0, time=21

Note: The speed difference can occur anywhere not necessarily on the 11 the second. within a minute, speed might be getting changed on 11th, 23rd, 44th sec etc.

Used a rule pipeline, one to calculate the records where speed was getting changed and another to evaluate the preceding 10 events, the event with the speed change and the following 10 events.

#1 Created the source stream
{"sql" : "create stream demo () WITH (DATASOURCE="devices/snapshot", FORMAT="JSON")"}

#2 Create rule to calculate the records where speed was changing and publish to another stream
{
"id": "rule1",
"sql": "SELECT * FROM demo WHERE speed-LAG(speed)!=0",
"actions": [{
"log": {
},
"memory": {
"topic": "check/memo"
}
}]
}

#3 Create a stream from the memory topic to get the preceding 10 events, and the following 10 events.
{"sql" : "create stream snap() WITH (DATASOURCE="check/memo", FORMAT="JSON", TYPE="memory")"}

#4 Create second rule
{
"id": "rule2",
"sql": "select * from demo LEFT JOIN snap ON demo.time=snap.timeWHERE demo.time BETWEEN(snap.time-10) AND (snap.time+10)
GROUP BY COUNTWINDOW(21,1)"
"actions": [{
"log": {
}
}]
}

However, rule 2 is not producing correct results. Can someone point me in the right direction?
i was also trying to come up with this rule attached in the screenshot as an alternative to rule2 for capturing preceding and following events. However, I am feeling lost here.
SELECT *
FROM demo
WHERE time BETWEEN
(SELECT time FROM snap) - 10
AND
(SELECT time FROM snap) + 10; // This produces an error.

[ngjaying]

Are you trying to capture events 10 seconds before and after a condition met? If so, we are actually working to extend sliding window to support such requirement. There are now 2 PRs under review #2005 & #2008. You may want to try after they are merged.

[Richapagare]

Hello @ngjaying , yes I am trying to capture events 10 seconds before and after a condition is met. For this I was wondering if we could do a rule pipeling and write the rules. Can we achieve this without using sliding window?

I was trying this but it produces an error.
SELECT *
FROM demo
WHERE time BETWEEN
(SELECT time FROM snap) - 10
AND
(SELECT time FROM snap) + 10;

[superrxan]

Hi @Richapagare
you can try this SQL

SELECT * from demo GROUP BY COUNTWINDOW(21,1) HAVING collect(speed)[10]-collect(speed)[9] != 0

[Yisaer]

Hi，I think we can capture the data by following sql:

select * from demo group by slidingWindow(ss,10,10) over (when speed_diff = 1);