StreamInsight，事件匹配

Question

在StreamInsight中，我遇到了一种情况，即我有一个输入源，其不同类型的事件需要不同的处理，但最终与来自同一源的其他事件相匹配。

我在下面创建了一个(非常)简单的场景，其中输入源生成随机数(为了简单起见，1和0)。如果号码是偶数，我们希望保存到进一步的通知(未知的持续时间)。如果数字是奇数，我们希望将它与偶数流中的n-1匹配，然后从偶数流中删除n-1。如果没有匹配，则将奇数简单地按原样处理，没有进一步的计算。我已经按照预期的方式工作到从偶数流中移除匹配的n-1。进行匹配，并将匹配推送到输出适配器，但对于要对给定事件进行另一个连接而言，仍然可用。我收集了几天的实验和研究成果，不知怎么的，我需要剪辑均匀流事件持续时间(ClipEventDuration)，这大概是GenerateEvenJoinQuery过滤器的一部分，但是，我尝试过的每一件事都没有产生任何变化，或者没有产生预期的结果。我还试着用更少的运气将evenStream更改为间隔形状。如有任何帮助或建议，将不胜感激。

例如，给定一个简化列表: 1，0，1，1，0，0，1，1，1

我希望输出结果如下: 1,100，1,100,100，1

我也会接受作为我正在使用的真实场景，第一个输出实际上是不可能的:注意，第2和第3 0是连接到一个单一的1.1,100，1,100，1，1，1

...
    CepStream<int> inputStream = CepStream<int>.Create(
            app
            , "ATGInputStream"
            , typeof(InputAdapterFactory)
            , new InputConfig()
            , EventShape.Point);


    var everythingFilter = from e in inputStream select e;
    Query everythingQuery = GenerateEverythingQuery(app, inputStream);
    var everythingStream = everythingQuery.ToStream<int>("everythingStream");

    Query oddQuery = GenerateOddQuery(app, everythingStream);
    var oddAts = new AdvanceTimeSettings(new AdvanceTimeGenerationSettings(1, TimeSpan.FromTicks(-1), false), null, AdvanceTimePolicy.Drop);
    var oddStream = oddQuery.ToStream<int>("oddStream", oddAts);

    // only inject a cti in to even when we need it
    var ats = new AdvanceTimeSettings(null, new AdvanceTimeImportSettings("oddStream"), AdvanceTimePolicy.Adjust);
    Query evenQuery = GenerateEvenQuery(app, everythingStream);
    var evenStream = evenQuery.ToStream<int>("evenStream", ats);

    Query joinQuery = GenerateOddEvenJoinQuery(app, evenStream, oddStream);
    var joinStream = joinQuery.ToStream<int>("joinStream");
...


private Query GenerateOddEvenJoinQuery(Application app, CepStream<int> evenStream, CepStream<int> oddStream) {
    // (o * e) + 100 is an easy way to tell we had a match
    var filter = (from o in oddStream
                  from e in evenStream
                  where e == (o - 1)
                  select (o * e) + 100);

    // LEFT ANTI SEMI JOIN
    var filter2 = from o in oddStream
                    where (from e in evenStream where e == o - 1 select e).IsEmpty()
                    select o;

    var joinFilter = filter.Union(filter2);

    return joinFilter.ToQuery(
            app
            , "Number Join Query"
            , "Joins number streams."
            , EventShape.Point
            , StreamEventOrder.FullyOrdered);
}

private Query GenerateEvenQuery(Application app, CepStream<int> stream) {
    var evenFilter = (from e in stream where e % 2 == 0 select e).AlterEventDuration(e => TimeSpan.MaxValue);

    return evenFilter.ToQuery(
            app
            , "EvenQuery"
            , ""
            , EventShape.Edge
            , StreamEventOrder.FullyOrdered);
}

private Query GenerateOddQuery(Application app, CepStream<int> stream) {
    var filter = (from f in stream where (f % 2) == 1 select f);

    return filter.ToQuery(
            app
            , "OddQuery"
            , "Queries for odd numbers in stream."
            , EventShape.Point
            , StreamEventOrder.FullyOrdered);
}

private Query GenerateEverythingQuery(Application app, CepStream<int> stream) {
    var everythingFilter = from e in stream select e;

    return everythingFilter.ToQuery(
            app
            , "EverythingQuery"
            , "Queries everything from the input stream."
            , EventShape.Point
            , StreamEventOrder.FullyOrdered);
}

解决办法：

虽然我希望有一些更精细和可能更快的东西，但延迟处理可能有助于提高性能。

public Program() {
    public Program() {
        ...
            var stream = CepStream<RandomNumber>.Create(
                    app
                    , "StaticInputStream"
                    , typeof(StaticInputAdapterFactory)
                    , new InputConfig()
                    , EventShape.Point);

            var processedStream = stream.Scan(new StreamMatcher());

            Query consoleQuery = GenerateConsoleOutputQuery(app, processedStream);
        ...
    }

    private Query GenerateConsoleOutputQuery(Application app, CepStream<int> stream) {
        var filter = from f in stream select f;

        return filter.ToQuery(
                app
                , "Console Output Query"
                , "Queries for messages to output to the console."
                , typeof(OutputAdapterFactory)
                , new OutputConfig()
                , EventShape.Point
                , StreamEventOrder.FullyOrdered);
    }

    public class StreamMatcher : CepPointStreamOperator<RandomNumber, int> {
        private List<int> unmatched = new List<int>();

        public override bool IsEmpty {
            get { return false; }
        }

        public override IEnumerable<int> ProcessEvent(PointEvent<RandomNumber> inputEvent) {
            if(inputEvent.Payload.value % 2 == 0) {
                unmatched.Add(inputEvent.Payload.value);
            } else {
                var result = inputEvent.Payload.value;

                int match = -1;

                try {
                    match = (from f in unmatched where f == result - 1 select f).Take(1).Single();
                    unmatched.Remove(match);
                } catch { }

                if(match > -1) {
                    result += match + 100;
                }

                yield return result;
            }
        }
    }
}

public class RandomNumber {
    public int value { get; set; }
    public DateTime timeStamp { get; set; }
}

Answer 1

您可以考虑使用UDSO (用户定义的流操作符)，其中可以保持零的状态。

void Main()
{
    var randomNumbers = new []
    {
        new RandomNumber(){ Value = 1, TimeStamp = DateTime.Parse("2012-01-01 10:01:00 AM")  },
        new RandomNumber(){ Value = 0, TimeStamp = DateTime.Parse("2012-01-01 10:02:00 AM")  },
        new RandomNumber(){ Value = 0, TimeStamp = DateTime.Parse("2012-01-01 10:02:00 AM")  },
        new RandomNumber(){ Value = 1, TimeStamp = DateTime.Parse("2012-01-01 10:03:00 AM")  },
        new RandomNumber(){ Value = 1, TimeStamp = DateTime.Parse("2012-01-01 10:04:00 AM")  },
        new RandomNumber(){ Value = 0, TimeStamp = DateTime.Parse("2012-01-01 10:05:00 AM")  },

    };

    var stream =  randomNumbers.ToPointStream(Application, 
            e=> PointEvent.CreateInsert(e.TimeStamp ,e), 
            AdvanceTimeSettings.IncreasingStartTime) ;

    var query = stream.Scan(new MyCalculation());

    query.Dump();   

}


public class MyCalculation : CepPointStreamOperator<RandomNumber,string>
{

    private Queue<int> _queue = new Queue<int>() ;

    public override bool IsEmpty
    {
        get { return false; }
    }

    public override IEnumerable<string> ProcessEvent(PointEvent<RandomNumber> inputEvent)
    {
        if (inputEvent.Payload.Value % 2 == 0)
        {
                _queue.Enqueue(inputEvent.Payload.Value);
        }
        else
        {
            var result= inputEvent.Payload.Value.ToString() ;
            var list = _queue.ToArray();
            for (int i = 0; i < list.Count(); i++)
            {
                result += list[i];
            }
             yield return result ;
        }
    }
}


public class RandomNumber 
{
    public int Value {get;set;}
    public DateTime TimeStamp {get;set;}
}