Esper 多个属性上的不同事件

Question

我对 Esper 中的流语义有疑问。我的目标是只输出具有成对不同属性的事件。此外，属性之间还存在一些时间条件（参见 Espers Interval Algebra Reference）。

示例语句：

insert into output_stream select a.*, b.*
from stream1#length(100) as a, stream2#length(100) as b
where a.before(b) or a.meets(b) or a.overlaps(b)

成对不同的属性意味着，我想确保没有两个输出 o1, o2 where o1.a = o2.a or o1.b = o2.b。举个更具体的例子，如果有结果

o1: (a = a1, b = b1),
o2: (a = a1, b = b2),
o3: (a = a2, b = b2),
o4: (a = a2, b = b1)

只应输出其中两个（例如 o1 和 o3 或 o2 和 o4）。哪个暂时不重要。

我想用这样的NOT EXISTS 子句来完成成对的不同属性：

NOT EXISTS ( 
    select * from output_stream#length(100) as otherOutput 
    where a = otherOutput.a or b = otherOutput.b )

这部分有效，对于连续输出，断言o1.a = o2.a or o1.b = o2.b 始终成立。

但是，当stream1 首先提供多个“a”，然后stream2 提供一个“b”，这与要与两个“a”连接的条件相匹配时，一次会有多个输出 。这不在我的NOT EXISTS 子句中，因为在同一步骤中会出现多个具有相同“b”的输出，因此它们还没有在output_stream 中。

distinct 关键字在这里不合适，因为它会同时检查所有属性，而不是成对检查。同样，所有属性上的简单group by 也不合适。我希望有类似“a 上不同，b 上不同”之类的标准，但它不存在。

我可以使用嵌套的group bys 来解决这个问题，我在每个属性上进行分组

select first(*) from (select first(*) from output_stream group by a) group by b

但根据一条评论，在流处理系统中没有明确定义的语义。因此，Esper 不允许在查询的 from 部分中使用子查询。

我需要的是一种强制一次只输出一个输出的方法，因此在每个进一步的输出上重新检查NOT EXISTS 条件，或者在实际插入之前以某种方式检查同时发生的输出将它们放入流中。

更新： 输出的时序不是很关键。 output_stream 将被其他此类语句使用，因此我可以通过增加窗口的长度来解决延迟问题。 stream1 和 stream2 按其 startTimestamp 属性的顺序传递事件。

Answer 1

create schema Pair(a string, b string);
create window PairWindow#length(100) as Pair;
insert into PairWindow select * from Pair;
on PairWindow as arriving select * from PairWindow as other  
  where arriving.a = other.a or arriving.b = other.b

这是一个使用命名窗口的示例自联接，该窗口保留最后 100 对。

编辑：上面的查询是为了我对原始要求的理解而设计的。下面的查询是为新的说明而设计的。它检查“a”或“b”是否有任何先前的值（在最后 100 个事件中，根据需要将 #length(100) 关闭）

create schema Pair(a string, b string);
create window PairUniqueByA#firstunique(a)#length(100) as Pair;
create window PairUniqueByB#firstunique(b)#length(100) as Pair;

insert into PairUniqueByA select * from Pair;
insert into PairUniqueByB select * from Pair;

select * from Pair as pair
  where not exists (select a from PairUniqueByA as uba where uba.a = pair.a)
  and not exists (select a from PairUniqueByB as ubb where ubb.b = pair.b);