在SiddhiQL中检测事件A重复发生的非窗口实现方案咨询

Great question—handling high-throughput event streams without choking on memory is a classic stream processing pain point. Let’s break down a few practical alternatives to windowing that’ll work for your use case:

1. Keyed State with Time-to-Live (TTL) #

Instead of storing a full window of events, only track the last occurrence of each Event A using Siddhi’s state management with automatic TTL cleanup. This way, you only keep minimal state (just the event identifier) and it gets purged automatically after 20 minutes.

Here’s a SiddhiQL example:

@App:name("EventADuplicateDetection")
@App:description("Detect repeat EventA within 20 mins using TTL-backed state")

define stream EventStream (eventId string, eventType string);

@sink(type='log', prefix='REPEATED EVENT ALERT:')
define stream RepeatedEventStream (eventId string);

-- Define a state store with 20-minute TTL (1200000 milliseconds)
@State(state.id='eventTracker', state.ttl='1200000')
define table EventTrackerState (eventId string);

-- Trigger alert if EventA already exists in state (i.e., seen in last 20 mins)
from EventStream[eventType == 'A'] as e
join EventTrackerState as s
    on e.eventId == s.eventId
select e.eventId
insert into RepeatedEventStream;

-- Add new EventA entries to state if they don't already exist
from EventStream[eventType == 'A'] as e
left join EventTrackerState as s
    on e.eventId == s.eventId
where s.eventId is null
select e.eventId
insert into EventTrackerState;

Why this works: You’re only storing the event IDs you need to track, not the full event payload or a sliding window of all events. The TTL ensures old entries are cleaned up automatically, preventing memory leaks.

2. Stateless Timestamp Comparison with External Storage #

For extreme throughput where even in-memory state is a concern, offload the state tracking to an external key-value store (like Redis or HBase). Each time Event A arrives, check the stored timestamp of its last occurrence to see if it falls within the 20-minute window.

Example with Redis integration in SiddhiQL:

@App:name("TimestampBasedEventDetection")

define stream EventStream (eventId string, eventType string, eventTimestamp long);

@sink(type='log', prefix='REPEATED EVENT ALERT:')
define stream RepeatedEventStream (eventId string);

-- Connect to Redis as the state store with 20-minute TTL
@Store(type='redis', host='localhost', port='6379', ttl='1200000')
define table EventTimestampStore (eventId string, lastSeenTimestamp long);

-- Check if EventA was seen in the last 20 minutes
from EventStream[eventType == 'A'] as e
join EventTimestampStore as s
    on e.eventId == s.eventId
where e.eventTimestamp - s.lastSeenTimestamp <= 1200000
select e.eventId
insert into RepeatedEventStream;

-- Update or insert the latest timestamp for EventA
from EventStream[eventType == 'A'] as e
left join EventTimestampStore as s
    on e.eventId == s.eventId
select e.eventId, e.eventTimestamp as lastSeenTimestamp
update or insert into EventTimestampStore
    on EventTimestampStore.eventId == e.eventId;

Why this works: Siddhi only handles the event processing logic, while the external store manages state. This eliminates in-memory pressure entirely, and most key-value stores support native TTL to auto-clean old entries.

3. Optimized Window (If You Must Use One) #

If you’re set on using a window, you can drastically reduce memory usage by only retaining the minimal data needed (event ID and timestamp) instead of full event payloads. Use a sliding time window with field projection to keep storage lean:

@App:name("LeanWindowDetection")

define stream EventStream (eventId string, eventType string, eventTimestamp long);

@sink(type='log', prefix='REPEATED EVENT ALERT:')
define stream RepeatedEventStream (eventId string);

-- Only keep eventId and timestamp in the window, not full event data
from EventStream[eventType == 'A'] as e1
join EventStream[eventType == 'A']#window.slidingTime(20 min).select(eventId, eventTimestamp) as e2
    on e1.eventId == e2.eventId and e1.eventTimestamp > e2.eventTimestamp
select distinct e1.eventId
insert into RepeatedEventStream;

Why this works: By projecting only the necessary fields, you cut down on the memory footprint of the window. It’s still less efficient than the state-based approaches, but better than a full payload window for high-throughput streams.

内容的提问来源于stack exchange，提问作者Vishal Gaikwad