0

嗨：我需要在 Siddhi 流处理中实现以下重复数据删除逻辑。假设我有一个 InputStream，我想生成 OutputStream，如下所示：

(1) 当事件是 InputStream 中的第一个事件时（自事件处理引擎启动以来），将事件插入到 OutputStream 中。

(2)如果具有相同签名的事件，例如相同的事件名称，在2分钟的窗口内到达，我们认为该事件是相同的，我们不应该将该事件插入到OutputStream中。否则，我们应该将事件插入到 OutputStream 中。

我尝试使用事件模式进行过滤。但是，我找不到我可以在 Siddhi 中表达“否定逻辑”，即 if (not (e1 --> e2 with same signature in 2 minute window))。有没有一种聪明的方法来执行这样的事件重复数据删除逻辑？请注意，事件重复数据删除是事件处理所需的非常常见的表达方式。

如果我用 Java 实现它，那是相对简单的。我将创建一个哈希表。当第一个事件到达时，我将它注册到哈希表中，并将这个注册事件的事件可接受时间设置为 2 分钟后。当下一个事件到来时，我查找哈希表，并将检索到的事件的可接受时间与我当前的事件时间进行比较，如果当前事件时间小于可接受时间，我不会将其视为输出事件。如果可能的话，我更喜欢在 Siddhi 的流处理查询中实现声明性解决方案，而不是 Java 实现。

4

1 回答 1

0

您可以使用内存表并实现它；请在下面找到示例；它与您使用 Java 的方法非常相似。

    define stream InputStream (event_id string, data string);  
    define stream OutputStream (event_id string, data string);  
    define table ProcessedEvents (event_id string);

    from InputStream[not(ProcessedEvents.event_id == event_id in ProcessedEvents)]  
    insert into OutputStream ; 

    from OutputStream  
    select event_id  
    insert into ProcessedEvents ; 

    from OutputStream#window.time(2 sec)  
    select event_id  
    insert expired events into PurgeStream ; 

    from PurgeStream  
    delete ProcessedEvents  
       on ProcessedEvents.event_id == event_id ;
于 2018-06-15T17:19:07.160 回答