1

我遇到了 StreamInsight 的情况，其中我有 1 个输入源，其中包含不同类型的事件，需要区别对待，但最终与来自同一源的其他事件匹配。

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

例如，给定一个简化列表： [ 1, 0, 1, 1, 0, 0, 1, 1, 1 ]

我希望输出看起来像： [ 1, 100, 1, 100, 100, 1 ]

我也会接受我正在使用的真实场景，第一个输出实际上是不可能的：注意第二个和第三个 0 连接到一个 1. [ 1, 100, 1, 100, 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; }
}
4

1 回答 1

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;}
}
于 2012-11-27T01:31:50.767 回答