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 ] のようになると思います。
私が取り組んでいる実際のシナリオとしても受け入れますが、最初の出力は実際には不可能です。2 番目と 3 番目の 0 が 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; }
}