

Remove Duplicate Events

Provides examples of removing duplicate events that arrive within a given time duration.

define stream TemperatureStream
        (sensorId string, seqNo string, temperature double);

@info(name = 'Deduplicate-sensorId')
from TemperatureStream#unique:deduplicate(sensorId, 1 min)
select *
insert into UniqueSensorStream;

@info(name = 'Deduplicate-sensorId-and-seqNo')
from TemperatureStream#unique:deduplicate(
                        str:concat(sensorId,'-',seqNo), 1 min)
select *
insert into UniqueSensorSeqNoStream;

define stream TemperatureStream (sensorId string, seqNo string, temperature double);
@info(name = 'Deduplicate-sensorId')
from TemperatureStream#unique:deduplicate(sensorId, 1 min)	Remove duplicate events arriving within 1 minute time gap, based on unique sensorId.
select * insert into UniqueSensorStream;
@info(name = 'Deduplicate-sensorId-and-seqNo')
from TemperatureStream#unique:deduplicate( str:concat(sensorId,'-',seqNo), 1 min)	Remove duplicate events arriving within 1 minute time gap, based on unique sensorId and seqNo combination.
select * insert into UniqueSensorSeqNoStream;

Behavior

When events are sent to TemperatureStream stream, following events will get emitted after deduplication on UniqueSensorStream stream via Deduplicate-sensorId query, and UniqueSensorSeqNoStream stream via Deduplicate-sensorId-and-seqNo query.

Time	Input to TemperatureStream	Output at UniqueSensorStream	Output at UniqueSensorSeqNoStream
9:00:00	['AD11', '200', 18.0]	['AD11', '200', 18.0]	['AD11', '200', 18.0]
9:00:10	['AD11', '201', 23.0]	-	['AD11', '201', 23.0]
9:00:20	['FR45', '500', 22.0]	['FR45', '500', 22.0]	['FR45', '500', 22.0]
9:00:40	['AD11', '200', 18.0]	-	-
9:00:50	['AD11', '202', 28.0]	-	['AD11', '202', 28.0]
9:01:05	['FR45', '501', 22.0]	-	['FR45', '501', 22.0]
9:01:10	['AD11', '203', 30.0]	['AD11', '203', 30.0]	['AD11', '203', 30.0]
9:02:20	['AD11', '202', 28.0]	['AD11', '202', 28.0]	['AD11', '202', 28.0]
9:03:10	['AD11', '204', 30.0]	-	['AD11', '204', 30.0]

Top