CEPStream 지연 및 이벤트 종료

Question

알려진 종료 시간이있는 간격 이벤트가 있다고 가정하면 해당 간격 이벤트가 만료되기 전에 다른 이벤트를 시작하는 방법이 있습니까? 이것의 목적은 그 시점에서 집계를해야한다는 것입니다.

또는 쿼리를 할 수있는 방법은 나의 이해는 더 나은 방법을 제안 해주십시오 완전히 꺼져 (이 원하는되지 않는하지만 난 그것에 의지해야 할 수도 있습니다) 경우

(from input in inputStream 
select input).AddDelay(input.EndTime - DateTime.UtcNow) 

있다.

Answer 1

먼저 소스 스트림을 필터링하여 x1 이벤트 스트림을 얻어야합니다.

var x1Stream = from e in sourceStream 
      where e.ItemId == "X1" 
      select new {e.ItemId, e.Timestamp} 

다음으로 비 x1 이벤트 스트림을 얻기 위해 원본 스트림을 필터링해야합니다.

var nonX1Stream = from e in sourceStream 
      where e.ItemId != "X1" 
      select e; 

이제 우리는 X1 이벤트 기간 동안 발생하는 모든 비 X1 이벤트의 목록을 얻을 수있는 비 X1 이벤트 스트림으로 X1 이벤트 스트림에 참여할 수 있습니다.

var x = from l in x1Stream 
    from r in nonX1Stream 
    select new {l.ItemId, l.Timestamp, r}; 

HoppingWindow의 어떤 필요 것 X1 이벤트 기간 동안 발생하는 비 X1 이벤트의 수를 얻는 것은 실제로는 일정 시간 동안 스트림의 이벤트를 카운트 할 수 있습니다. ToEnumerable()을 호출하여 창없이 그룹화를 수행 할 수도 있습니다.

var y = from e in x.ToEnumerable() 
    group e by new {e.ItemId, e.Timestamp} 
    into g 
    select new {g.Key.ItemId, g.Key.Timestamp, Count = g.Count()};