I would suggest you using transaction command if the data volume is not so high. The biggest advantage is that it enables you to aggregate similar events from the distinct sources in one transaction while providing a "duration" field based on the _time used between the similar events.
By using eval's mvindex() you are then able to keep only the last or first events from the transaction.
... View more