StreamInsight Team : LINQ

Using Subjects to Deploy Queries Dynamically

Roman Schindlauer — Thu, 30 Aug 2012 18:13:00 GMT

In the previous blog posting, we showed how to construct and deploy query fragments to a StreamInsight server, and how to re-use them later. In today’s posting we’ll integrate this pattern into a method of dynamically composing a new query with an existing one.

The construct that enables this scenario in StreamInsight V2.1 is a Subject. A Subject lets me create a junction element in an existing query that I can tap into while the query is running.

To set this up as an end-to-end example, let’s first define a stream simulator as our data source:

var generator = myApp.DefineObservable(
(TimeSpan t) => Observable.Interval(t).Select(_ => new SourcePayload()));

This ‘generator’ produces a new instance of SourcePayload with a period of t (system time) as an IObservable. SourcePayload happens to have a property of type double as its payload data.

Let’s also define a sink for our example—an IObserver of double values that writes to the console:

var console = myApp.DefineObserver(
(string label) => Observer.Create<double>(e => Console.WriteLine("{0}: {1}", label, e)))
.Deploy("ConsoleSink");

The observer takes a string as parameter which is used as a label on the console, so that we can distinguish the output of different sink instances. Note that we also deploy this observer, so that we can retrieve it later from the server from a different process.

Remember how we defined the aggregation as an IQStreamable function in the previous article? We will use that as well:

var avg = myApp
    .DefineStreamable((IQStreamable<SourcePayload> s, TimeSpan w) =>
        from win in s.TumblingWindow(w)
        select win.Avg(e => e.Value))
    .Deploy("AverageQuery");

Then we define the Subject, which acts as an observable sequence as well as an observer. Thus, we can feed a single source into the Subject and have multiple consumers—that can come and go at runtime—on the other side:

var subject = myApp.CreateSubject("Subject", () => new Subject<SourcePayload>());

Subject are always deployed automatically. Their name is used to retrieve them from a (potentially) different process (see below).

Note that the Subject as we defined it here doesn’t know anything about temporal streams. It is merely a sequence of SourcePayloads, without any notion of StreamInsight point events or CTIs. So in order to compose a temporal query on top of the Subject, we need to 'promote' the sequence of SourcePayloads into an IQStreamable of point events, including CTIs:

var stream = subject.ToPointStreamable(
e => PointEvent.CreateInsert<SourcePayload>(e.Timestamp, e),
AdvanceTimeSettings.StrictlyIncreasingStartTime);

In a later posting we will show how to use Subjects that have more awareness of time and can be used as a junction between QStreamables instead of IQbservables.

Having turned the Subject into a temporal stream, we can now define the aggregate on this stream. We will use the IQStreamable entity avg that we defined above:

var longAverages = avg(stream, TimeSpan.FromSeconds(5));

In order to run the query, we need to bind it to a sink, and bind the subject to the source:

var standardQuery = longAverages
.Bind(console("5sec average"))
.With(generator(TimeSpan.FromMilliseconds(300)).Bind(subject));

Lastly, we start the process:

standardQuery.Run("StandardProcess");

Now we have a simple query running end-to-end, producing results. What follows next is the crucial part of tapping into the Subject and adding another query that runs in parallel, using the same query definition (the “AverageQuery”) but with a different window length. We are assuming that we connected to the same StreamInsight server from a different process or even client, and thus have to retrieve the previously deployed entities through their names:

// simulate the addition of a 'fast' query from a separate server connection,
// by retrieving the aggregation query fragment
// (instead of simply using the 'avg' object)
var averageQuery = myApp
    .GetStreamable<IQStreamable<SourcePayload>, TimeSpan, double>("AverageQuery");

// retrieve the input sequence as a subject
var inputSequence = myApp
    .GetSubject<SourcePayload, SourcePayload>("Subject");

// retrieve the registered sink
var sink = myApp.GetObserver<string, double>("ConsoleSink");

// turn the sequence into a temporal stream
var stream2 = inputSequence.ToPointStreamable(
    e => PointEvent.CreateInsert<SourcePayload>(e.Timestamp, e),
    AdvanceTimeSettings.StrictlyIncreasingStartTime);

// apply the query, now with a different window length
var shortAverages = averageQuery(stream2, TimeSpan.FromSeconds(1));

// bind new sink to query and run it
var fastQuery = shortAverages
    .Bind(sink("1sec average"))
    .Run("FastProcess");

The attached solution demonstrates the sample end-to-end.

Regards,
The StreamInsight Team

Taming Hopping Windows

Roman Schindlauer — Wed, 04 Jul 2012 00:03:06 GMT

At first glance, hopping windows seem fairly innocuous and obvious. They organize events into windows with a simple periodic definition: the windows have some duration d (e.g. a window covers 5 second time intervals), an interval or period p (e.g. a new window starts every 2 seconds) and an alignment a (e.g. one of those windows starts at 12:00 PM on March 15, 2012 UTC).

var wins = xs
    .HoppingWindow(TimeSpan.FromSeconds(5),
                   TimeSpan.FromSeconds(2),
                   new DateTime(2012, 3, 15, 12, 0, 0, DateTimeKind.Utc));

Logically, there is a window with start time a + np and end time a + np + d for every integer n. That’s a lot of windows. So why doesn’t the following query (always) blow up?

var query = wins.Select(win => win.Count());

A few users have asked why StreamInsight doesn’t produce output for empty windows. Primarily it’s because there is an infinite number of empty windows! (Actually, StreamInsight uses DateTimeOffset.MaxValue to approximate “the end of time” and DateTimeOffset.MinValue to approximate “the beginning of time”, so the number of windows is lower in practice.)

That was the good news. Now the bad news. Events also have duration. Consider the following simple input:

var xs = this.Application
                .DefineEnumerable(() => new[]
                    { EdgeEvent.CreateStart(DateTimeOffset.UtcNow, 0) })
                .ToStreamable(AdvanceTimeSettings.IncreasingStartTime);

Because the event has no explicit end edge, it lasts until the end of time. So there are lots of non-empty windows if we apply a hopping window to that single event! For this reason, we need to be careful with hopping window queries in StreamInsight. Or we can switch to a custom implementation of hopping windows that doesn’t suffer from this shortcoming.

The alternate window implementation produces output only when the input changes. We start by breaking up the timeline into non-overlapping intervals assigned to each window. In figure 1, six hopping windows (“Windows”) are assigned to six intervals (“Assignments”) in the timeline. Next we take input events (“Events”) and alter their lifetimes (“Altered Events”) so that they cover the intervals of the windows they intersect. In figure 1, you can see that the first event e₁ intersects windows w₁ and w₂ so it is adjusted to cover assignments a₁ and a₂. Finally, we can use snapshot windows (“Snapshots”) to produce output for the hopping windows. Notice however that instead of having six windows generating output, we have only four. The first and second snapshots correspond to the first and second hopping windows. The remaining snapshots however cover two hopping windows each! While in this example we saved only two events, the savings can be more significant when the ratio of event duration to window duration is higher.

Figure 1: Timeline

The implementation of this strategy is straightforward. We need to set the start times of events to the start time of the interval assigned to the earliest window including the start time. Similarly, we need to modify the end times of events to the end time of the interval assigned to the latest window including the end time. The following snap-to-boundary function that rounds a timestamp value t down to the nearest value t' <= t such that t' is a + np for some integer n will be useful. For convenience, we will represent both DateTime and TimeSpan values using long ticks:

static long SnapToBoundary(long t, long a, long p)
{
return t - ((t - a) % p) - (t > a ? 0L : p);
}

How do we find the earliest window including the start time for an event? It’s the window following the last window that does not include the start time assuming that there are no gaps in the windows (i.e. duration < interval), and limitation of this solution. To find the end time of that antecedent window, we need to know the alignment of window ends:

long e = a + (d % p);

Using the window end alignment, we are finally ready to describe the start time selector:

static long AdjustStartTime(long t, long e, long p)
{
return SnapToBoundary(t, e, p) + p;
}

To find the latest window including the end time for an event, we look for the last window start time (non-inclusive):

public static long AdjustEndTime(long t, long a, long d, long p)
{
return SnapToBoundary(t - 1, a, p) + p + d;
}

Bringing it together, we can define the translation from events to ‘altered events’ as in Figure 1:

public static IQStreamable<T> SnapToWindowIntervals<T>(IQStreamable<T> source, TimeSpan duration, TimeSpan interval, DateTime alignment)
{
    if (source == null) throw new ArgumentNullException("source");

    // reason about DateTime and TimeSpan in ticks
    long d = Math.Min(DateTime.MaxValue.Ticks, duration.Ticks);
    long p = Math.Min(DateTime.MaxValue.Ticks, Math.Abs(interval.Ticks));

    // set alignment to earliest possible window
    var a = alignment.ToUniversalTime().Ticks % p;

    // verify constraints of this solution
    if (d <= 0L) { throw new ArgumentOutOfRangeException("duration"); }
    if (p == 0L || p > d) { throw new ArgumentOutOfRangeException("interval"); }

    // find the alignment of window ends
    long e = a + (d % p);

    return source.AlterEventLifetime(
        evt => ToDateTime(AdjustStartTime(evt.StartTime.ToUniversalTime().Ticks, e, p)),
        evt => ToDateTime(AdjustEndTime(evt.EndTime.ToUniversalTime().Ticks, a, d, p)) -
            ToDateTime(AdjustStartTime(evt.StartTime.ToUniversalTime().Ticks, e, p)));
}

public static DateTime ToDateTime(long ticks)
{
    // just snap to min or max value rather than under/overflowing
    return ticks < DateTime.MinValue.Ticks
        ? new DateTime(DateTime.MinValue.Ticks, DateTimeKind.Utc)
        : ticks > DateTime.MaxValue.Ticks
        ? new DateTime(DateTime.MaxValue.Ticks, DateTimeKind.Utc)
        : new DateTime(ticks, DateTimeKind.Utc);
}

Finally, we can describe our custom hopping window operator:

public static IQWindowedStreamable<T> HoppingWindow2<T>(
    IQStreamable<T> source,
    TimeSpan duration,
    TimeSpan interval,
    DateTime alignment)
{
    if (source == null) { throw new ArgumentNullException("source"); }
    return SnapToWindowIntervals(source, duration, interval, alignment).SnapshotWindow();
}

By switching from HoppingWindow to HoppingWindow2 in the following example, the query returns quickly rather than gobbling resources and ultimately failing!

public void Main()
{
    var start = new DateTimeOffset(new DateTime(2012, 6, 28), TimeSpan.Zero);
    var duration = TimeSpan.FromSeconds(5);
    var interval = TimeSpan.FromSeconds(2);
    var alignment = new DateTime(2012, 3, 15, 12, 0, 0, DateTimeKind.Utc);

    var events = this.Application.DefineEnumerable(() => new[]
    {
        EdgeEvent.CreateStart(start.AddSeconds(0), "e0"),
        EdgeEvent.CreateStart(start.AddSeconds(1), "e1"),
        EdgeEvent.CreateEnd(start.AddSeconds(1), start.AddSeconds(2), "e1"),
        EdgeEvent.CreateStart(start.AddSeconds(3), "e2"),
        EdgeEvent.CreateStart(start.AddSeconds(9), "e3"),
        EdgeEvent.CreateEnd(start.AddSeconds(3), start.AddSeconds(10), "e2"),
        EdgeEvent.CreateEnd(start.AddSeconds(9), start.AddSeconds(10), "e3"),
    }).ToStreamable(AdvanceTimeSettings.IncreasingStartTime);

    var adjustedEvents = SnapToWindowIntervals(events, duration, interval, alignment);

    var query = from win in HoppingWindow2(events, duration, interval, alignment)
                select win.Count();

    DisplayResults(adjustedEvents, "Adjusted Events");
    DisplayResults(query, "Query");
}

As you can see, instead of producing a massive number of windows for the open start edge e₀, a single window is emitted from 12:00:15 AM until the end of time:

Adjusted Events

StartTime	EndTime	Payload
6/28/2012 12:00:01 AM	12/31/9999 11:59:59 PM	e0
6/28/2012 12:00:03 AM	6/28/2012 12:00:07 AM	e1
6/28/2012 12:00:05 AM	6/28/2012 12:00:15 AM	e2
6/28/2012 12:00:11 AM	6/28/2012 12:00:15 AM	e3

Query

StartTime	EndTime	Payload
6/28/2012 12:00:01 AM	6/28/2012 12:00:03 AM	1
6/28/2012 12:00:03 AM	6/28/2012 12:00:05 AM	2
6/28/2012 12:00:05 AM	6/28/2012 12:00:07 AM	3
6/28/2012 12:00:07 AM	6/28/2012 12:00:11 AM	2
6/28/2012 12:00:11 AM	6/28/2012 12:00:15 AM	3
6/28/2012 12:00:15 AM	12/31/9999 11:59:59 PM	1

Regards,
The StreamInsight Team

Changes to the LINQ-to-StreamInsight Dialect

Roman Schindlauer — Thu, 28 Jun 2012 17:09:00 GMT

In previous versions of StreamInsight (1.0 through 2.0), CepStream<> represents temporal streams of many varieties:

Streams with ‘open’ inputs (e.g., those defined and composed over CepStream<T>.Create(string streamName)
Streams with ‘partially bound’ inputs (e.g., those defined and composed over CepStream<T>.Create(Type adapterFactory, …))
Streams with fully bound inputs (e.g., those defined and composed over To*Stream – sequences or DQC)

The stream may be embedded (where Server.Create is used)
The stream may be remote (where Server.Connect is used)

When adding support for new programming primitives in StreamInsight 2.1, we faced a choice: Add a fourth variety (use CepStream<> to represent streams that are bound the new programming model constructs), or introduce a separate type that represents temporal streams in the new user model. We opted for the latter. Introducing a new type has the effect of reducing the number of (confusing) runtime failures due to inappropriate uses of CepStream<> instances in the incorrect context. The new types are:

IStreamable<>, which logically represents a temporal stream.
IQStreamable<> : IStreamable<>, which represents a queryable temporal stream. Its relationship to IStreamable<> is analogous to the relationship of IQueryable<> to IEnumerable<>. The developer can compose temporal queries over remote stream sources using this type.

The syntax of temporal queries composed over IQStreamable<> is mostly consistent with the syntax of our existing CepStream<>-based LINQ provider. However, we have taken the opportunity to refine certain aspects of the language surface. Differences are outlined below. Because 2.1 introduces new types to represent temporal queries, the changes outlined in this post do no impact existing StreamInsight applications using the existing types!

SelectMany

StreamInsight does not support the SelectMany operator in its usual form (which is analogous to SQL’s “CROSS APPLY” operator):

static IEnumerable<R> SelectMany<T, R>(this IEnumerable<T> source, Func<T, IEnumerable<R>> collectionSelector)

It instead uses SelectMany as a convenient syntactic representation of an inner join. The parameter to the selector function is thus unavailable. Because the parameter isn’t supported, its type in StreamInsight 1.0 – 2.0 wasn’t carefully scrutinized. Unfortunately, the type chosen for the parameter is nonsensical to LINQ programmers:

static CepStream<R> SelectMany<T, R>(this CepStream<T> source, Expression<Func<CepStream<T>, CepStream<R>>> streamSelector)

Using Unit as the type for the parameter accurately reflects the StreamInsight’s capabilities:

static IQStreamable<R> SelectMany<T, R>(this IQStreamable<T> source, Expression<Func<Unit, IQStreamable<R>>> streamSelector)

For queries that succeed – that is, queries that do not reference the stream selector parameter – there is no difference between the code written for the two overloads:

from x in xs
from y in ys
select f(x, y)

Top-K

The Take operator used in StreamInsight causes confusion for LINQ programmers because it is applied to the (unbounded) stream rather than the (bounded) window, suggesting that the query as a whole will return k rows:

(from win in xs.SnapshotWindow()
from x in win
orderby x.A
select x.B).Take(k)

The use of SelectMany is also unfortunate in this context because it implies the availability of the window parameter within the remainder of the comprehension. The following compiles but fails at runtime:

(from win in xs.SnapshotWindow()
from x in win
orderby x.A
select win).Take(k)

The Take operator in 2.1 is applied to the window rather than the stream:

Before	After
(from win in xs.SnapshotWindow() from x in win orderby x.A select x.B).Take(k)	from win in xs.SnapshotWindow() from b in (from x in win orderby x.A select x.B).Take(k) select b

Multicast

We are introducing an explicit multicast operator in order to preserve expression identity, which is important given the semantics about moving code to and from StreamInsight. This also better matches existing LINQ dialects, such as Reactive. This pattern enables expressing multicasting in two ways:

Implicit	Explicit
var ys = from x in xs where x.A > 1 select x; var zs = from y1 in ys from y2 in ys.ShiftEventTime(_ => TimeSpan.FromSeconds(1)) select y1 + y2;	var ys = from x in xs where x.A > 1 select x; var zs = ys.Multicast(ys1 => from y1 in ys1 from y2 in ys1.ShiftEventTime(_ => TimeSpan.FromSeconds(1)) select y1 + y2;

Notice the product translates an expression using implicit multicast into an expression using the explicit multicast operator. The user does not see this translation.

Default window policies

Only default window policies are supported in the new surface. Other policies can be simulated by using AlterEventLifetime.

Before	After
xs.SnapshotWindow( WindowInputPolicy.ClipToWindow, SnapshotWindowInputPolicy.Clip)	xs.SnapshotWindow()
xs.TumblingWindow( TimeSpan.FromSeconds(1), HoppingWindowOutputPolicy.PointAlignToWindowEnd)	xs.TumblingWindow( TimeSpan.FromSeconds(1))
xs.TumblingWindow( TimeSpan.FromSeconds(1), HoppingWindowOutputPolicy.ClipToWindowEnd)	Not supported
…

LeftAntiJoin

Representation of LASJ as a correlated sub-query in the LINQ surface is problematic as the StreamInsight engine does not support correlated sub-queries (see discussion of SelectMany). The current syntax requires the introduction of an otherwise unsupported ‘IsEmpty()’ operator. As a result, the pattern is not discoverable and implies capabilities not present in the server. The direct representation of LASJ is used instead:

Before	After
from x in xs where (from y in ys where x.A > y.B select y).IsEmpty() select x	xs.LeftAntiJoin(ys, (x, y) => x.A > y.B)
from x in xs where (from y in ys where x.A == y.B select y).IsEmpty() select x	xs.LeftAntiJoin(ys, x => x.A, y => y.B)

ApplyWithUnion

The ApplyWithUnion methods have been deprecated since their signatures are redundant given the standard SelectMany overloads:

Before	After
xs.GroupBy(x => x.A).ApplyWithUnion(gs => from win in gs.SnapshotWindow() select win.Count())	xs.GroupBy(x => x.A).SelectMany( gs => from win in gs.SnapshotWindow() select win.Count())
xs.GroupBy(x => x.A).ApplyWithUnion(gs => from win in gs.SnapshotWindow() select win.Count(), r => new { r.Key, Count = r.Payload })	from x in xs group x by x.A into gs from win in gs.SnapshotWindow() select new { gs.Key, Count = win.Count() }

Alternate UDO syntax

The representation of UDOs in the StreamInsight LINQ dialect confuses cardinalities. Based on the semantics of user-defined operators in StreamInsight, one would expect to construct queries in the following form:

from win in xs.SnapshotWindow()
from y in MyUdo(win)
select y

Instead, the UDO proxy method is referenced within a projection, and the (many) results returned by the user code are automatically flattened into a stream:

from win in xs.SnapshotWindow()
select MyUdo(win)

The “many-or-one” confusion is exemplified by the following example that compiles but fails at runtime:

from win in xs.SnapshotWindow()
select MyUdo(win) + win.Count()

The above query must fail because the UDO is in fact returning many values per window while the count aggregate is returning one.

Original syntax	New alternate syntax
from win in xs.SnapshotWindow() select win.UdoProxy(1)	from win in xs.SnapshotWindow() from y in win.UserDefinedOperator(() => new Udo(1)) select y -or- from win in xs.SnapshotWindow() from y in win.UdoMacro(1) select y

Notice that this formulation also sidesteps the dynamic type pitfalls of the existing “proxy method” approach to UDOs, in which the type of the UDO implementation (TInput, TOuput) and the type of its constructor arguments (TConfig) need to align in a precise and non-obvious way with the argument and return types for the corresponding proxy method.

UDSO syntax

UDSO currently leverages the DataContractSerializer to clone initial state for logical instances of the user operator. Initial state will instead be described by an expression in the new LINQ surface.

Before	After
xs.Scan(new Udso())	xs.Scan(() => new Udso())

Name changes

ShiftEventTime => AlterEventStartTime: The alter event lifetime overload taking a new start time value has been renamed.

CountByStartTimeWindow => CountWindow

New in V1.2: Default Window Output Policies

Roman Schindlauer — Wed, 03 Aug 2011 14:46:00 GMT

Who knows what window output policies in StreamInsight LINQ are? Anyone? Well, you have been using them whenever you have used a window. Output policies specify the timestamps of the result of the aggregation that happens on the window. Even if you didn’t really care about that, you had to specify the output policy parameter for the window. In V1.2 we are introducing

Default policies so that you don’t have to specify them for every window, and
Another output policy for Hopping/Tumbling windows.

Importantly, there is no breaking change. Whatever you used in 1.1 is still working in 1.2, with the same behavior. For count- and snapshot-windows, the single existing output policy is becoming the default:

SnapshotWindow() is equivalent to
SnapshotWindow(SnapshotWindowOutputPolicy.Clip)

CountByStartTimeWindow(num) is equivalent to
CountByStartTimeWindow(num, CountWindowOutputPolicy.PointAlignToWindowEnd)

Set-based operations on top of snapshot windows always produce an event with the timestamps of the snapshot window itself, while operations on count windows produce a point event at the end of the window.

The only existing output policy for hopping and tumbling windows in StreamInsight 1.1 was to produce the window size as result event lifetime. Now this turned out to be impractical in many cases, especially with overlapping hopping windows. Have a look at the following diagram:

Just like the windows, also the result events overlap. When using this result in a subsequent operation, like a join with another stream, this easily leads to unexpected results. In many cases, only a single aggregation result at each point in time is needed (instead of four concurrently valid payloads in the diagram above). To this end, we have introduced another output policy for hopping (and tumbling) windows, which produces a point event at the end of the window:

This policy is now also the default for hopping windows:

HoppingWindow(size, hop) is equivalent to
HoppingWindow(size, hop, HoppingWindowOutputPolicy.PointAlignToWindowEnd)

TumblingWindow(size) is equivalent to
TumblingWindow(size, HoppingWindowOutputPolicy.PointAlignToWindowEnd)

This can now easily be converted into a continuous signal, for instance by using the ToSignal() macro that we discussed earlier. This signal has a payload at each point in time and hence is suitable to be joined with another stream, which is not necessarily synchronized in time. Typical example: join the result of a hopping aggregation with some reference stream:

var result = from a in avg.ToSignal(e => e.SourceId)
from r in reference.ToSignal(e => e.SourceId)
select new { alarm = a.Value < r.Threshold };

Dealing with point events instead of overlapping intervals in the context of windows helps avoiding unexpected side-effects and—most of the times—corresponds to the intended semantics.

Regards,
The StreamInsight Team

New in V1.2: LINQ Macros

Roman Schindlauer — Thu, 28 Jul 2011 15:54:00 GMT

StreamInsight LINQ is powerful and expressive. At the same time, the learning curve for a new user can be steep, because the query operators work on a semantically basic level. This means that complex use cases need to be build bottom up, composing the built-in operators to create higher-level query modules, etc. Many such modules are used over and over again, like the conversion of a stream to a signal, or the left-outer join. In StreamInsight V1.1 it was possible to some extent to write your own extension methods over CepStreams, but due to the mechanics and capabilities of the StreamInsight LINQ provider, for many such methods this was often difficult if not impossible. In V1.2 We overhauled the LINQ infrastructure significantly to make the definition of LINQ macros much easier. This enables you to express common building blocks as extension methods just once, and then use and re-use them to compose more sophisticated computations. The user of these macros doesn’t need to think in terms of AlterLifeTime or Left-Anti-Semi-Join anymore.

Let’s look at the first example, creating an event stream into a series of continuous, non-overlapping interval events. This is a typical preparation to implement a correlation (join) of two streams. In StreamInsight we can easily do this by extending the lifetime of each event and then clipping it right at the beginning of the next event (this is also described in the section “Point to Signal Conversion” on our MSDN pages):

var result = source
.AlterEventDuration(e => TimeSpan.MaxValue)
.ClipEventDuration(source, (s, e) => true);

Being a typical query pattern, I can easily convert this into an extension method:

public static CepStream<T> ToSignal<T>(this CepStream<T> inputstream)
{
    return inputstream
        .AlterEventDuration(e => TimeSpan.MaxValue)
        .ClipEventDuration(inputstream, (e1, e2) => true);
}

And use it as follows:

var result = source.ToSignal();

Now assume that my input is a multiplexed stream, containing events from different sources, identified by some payload field. Obviously, I want to convert each source into a separate signal, so the matching condition in ClipEventDuration() needs to change:

var result = source
.AlterEventDuration(e => TimeSpan.MaxValue)
.ClipEventDuration(source, (s, e) => s.Id == e.Id);

Ideally, I have an according extension method that allows me to pass the matching expression as a parameter:

var result = source.ToSignal((s, e) => s.Id == e.Id);

Here is how I can write such an extension method:

public static CepStream<T> ToSignal<T>(this CepStream<T> inputstream, Expression<Func<T, T, bool>> matchExpression)
{
    return inputstream
        .AlterEventDuration(e => TimeSpan.MaxValue)
        .ClipEventDuration(inputstream, (e1, e2) => matchExpression.Compile()(e1, e2));
}

The ability to compile expressions into delegates is new in V1.2. It also allows to create the equality predicate inside the macro:

public static CepStream<T> ToSignal<T, K>(this CepStream<T> inputstream, Expression<Func<T, K>> keySelector)
{
    return inputstream
        .AlterEventDuration(e => TimeSpan.MaxValue)
        .ClipEventDuration(inputstream, (e1, e2) => (keySelector.Compile()(e1)).Equals(keySelector.Compile()(e2)));
}

Now I can use the macro by just referring to the field that distinguishes my substreams:

var result = source.ToSignal(s => s.Id);

Our favorite example is the left outer join, which is not a native StreamInsight operator. In this operation, we want to join left and right streams, but also keep every event on the left side that does not join with the right side. We can model this as a plain inner join, a left-anti-semi-join (everything left without right), and the union thereof. Since Colin has already blogged about this pattern, I’ll not repeat it here and just refer to his posting.

Here is another really nice one: De-duplication. The following extension method helps you to get rid of duplicate events (same time, same payload):

public static CepStream<T> Distinct<T>(this CepStream<T> source)
{
    return from x in source
            group x by x into g
            from win in g.SnapshotWindow()
            select new { g.Key, IgnoreMe = win.Count() } into x
            select x.Key;
}

The usage now becomes very simple:

var result = input.Distinct();

Macros make it possible to build a domain-specific set of subqueries on top of the StreamInsight LINQ language, so that defining computations and analytics becomes more accessible and straightforward. The new V1.2 LINQPad samples include several other useful macros. Note that LINQPad does not allow true extension methods in the same file – they would have to be compiled and the resulting DLL referenced. This is why in our LINQPad samples the macros are defined as regular old methods.

Regards,
The StreamInsight Team

New in V1.2: Primitive Event Types

Roman Schindlauer — Mon, 25 Jul 2011 23:34:00 GMT

In a previous posting, we introduced nested types as one of the new features in StreamInsight V1.2. The second improvement in the type system represents the other end of the complexity spectrum, namely the ability to use "primitive" event types, without any struct or class. Remember the LINQ statement to compute the distances between two vehicles in that posting:

var distance = from a in vehicleA
              from b in vehicleB
              select new
              {
                  distance = Math.Sqrt(
                      Math.Pow(a.Loc.X – b.Loc.X, 2) +
                      Math.Pow(a.Loc.Y – b.Loc.Y, 2))
              };

Wouldn’t it be nice to be able to write this as:

var distance = from a in vehicleA
              from b in vehicleB
              select Math.Sqrt(
                  Math.Pow(a.Loc.X – b.Loc.X, 2) +
                  Math.Pow(a.Loc.Y – b.Loc.Y, 2));

Yes, it would – and in V1.2 you can! The type of this is now simply a CepStream<double>, and you can use it like this:

var result = from d in distance
where d > 9
select d;

Beauty! But wait, there is more: in V1.2, we also enabled the LINQ Let clause. We can consolidate the above statements nicely into a single one:

var withLet = from r in vehicleA
             from l in vehicleB
             let d = Math.Sqrt(
                 Math.Pow(r.Loc.X - l.Loc.X, 2) +
                 Math.Pow(r.Loc.Y - l.Loc.Y, 2))
             where d > 9
             select d;

You can find this example attached to run in LINQPad.

See how we can create a very concise StreamInsight sample over a stream of int in LINQPad:

var source = Enumerable.Range(1, 100).ToPointStream(
                Application,
                ev => PointEvent.CreateInsert(
                    DateTimeOffset.Now.AddMinutes(ev), ev),
                AdvanceTimeSettings.StrictlyIncreasingStartTime);

var result = from win in source.TumblingWindow(TimeSpan.FromMinutes(5))
            select win.Avg(e => e);

result.ToPointEnumerable().Where(e => e.EventKind == EventKind.Insert).Dump();

Alright, this query doesn’t really make a lot of sense, but you get the idea. This sample is also included in the set of new LINQPad samples – just update your LINQPad driver to get them.

Regards,
The StreamInsight Team

New in V1.2: Nested Event Types

Roman Schindlauer — Thu, 21 Jul 2011 16:50:00 GMT

One of the focus areas for us in V1.2 was the LINQ surface of StreamInsight. We have straightened out many of the kinks people complained about in previous versions and have added new features. Too many improvements for a single blog post, so today we’ll start with new features in the event type system.

One of the big limitations in StreamInsight so far was the requirement to express event types as a “flat” list of fields. What we are introducing with our current version is the ability to use nested structures as event types. You can now use fields that are instances of another class or struct, and you can define such nestings with arbitrary depth. Let’s look at an example:

public class Location
{
   public double X { get; set; }
   public double Y { get; set; }
}

public class Vehicle
{
   public string Id { get; set; }
   public Location Loc { get; set; }
}

Here, we are planning to use Vehicle as our event type, with a field of type Location. We can create our payloads in the input adapter (or as a .NET sequence item) as expected:

new Vehicle { Id = "A", Loc = new Location() { X = 2, Y = 3 } }

Within LINQ query statements, you can refer to these nested fields:

(We are assuming that these two input streams actually contain overlapping events, otherwise the join wouldn’t produce anything.)

Now we’d like to clarify a few things here:

The atomic fields in the structures still need to be of one of the StreamInsight-supported types. Collections as field types are not supported yet.
Nested structures always need to contain actual instances, they cannot be null.

How do you deal with nested fields in untyped adapters? Well, no change here. You still use the GetField/SetField methods of the event, passing the ordinal of the desired field. You get the ordinal for a field from its name through the CepEventType dictionary. The name for a nested field is for example Loc.X, just like in a programmatic reference.

Attached is a LINQPad sample that demonstrates nested types – you will also find it as part of the samples that come with the plugin. The sample also uses a macro for a common streaming design pattern, more about that in a later post. Don’t forget to update the plugin, btw!

Regards,
The StreamInsight Team

Rolling Out 101 Samples

Roman Schindlauer — Tue, 10 May 2011 22:33:00 GMT

StreamInsight’s LINQ query and analytics language is powerful and versatile. To ease users and developers over the inevitable initial learning curve, we know that high-quality documentation and a variety of examples are absolutely crucial. While we are trying to improve the former with every new release, we have also maintained a set of samples of the StreamInsight CodePlex site. Since the release of our plugin for the popular LINQPad, publishing query samples and snippets has become much easier and straightforward. We recently started collecting typical streaming computation patterns and expressing them in StreamInsight LINQ as basic building blocks – sort of the “101 StreamInsight Query Patterns”. Today we release the first set of these patterns, downloadable directly through the LINQPad UI:

Clicking on “Download more samples” will bring up a dialog window. Scrolling down you’ll see the StreamInsight sample section with a link to download:

This download will pick up the new samples, regardless of whether you already had a previous version or not.

Regards,
The StreamInsight Team

StreamInsight LINQPad Driver

Roman Schindlauer — Thu, 25 Nov 2010 01:03:00 GMT

Joseph Albahari's LINQPad has become an indispensable tool for many of us on the StreamInsight team. It allows you to quickly design, test and visualize results for LINQ queries. We've now developed a dedicated LINQPad driver for StreamInsight with Joe's help that you can use to quickly explore temporal data and StreamInsight's temporal LINQ dialect!

To get started, install .NET 4.0, LINQPad 4.0 and StreamInsight 1.1. You'll need the full version of StreamInsight (StreamInsight.msi, not StreamInsightClient.msi) and you'll need to create a server instance during setup. This version of the LINQPad driver evaluates stream queries locally with an embedded StreamInsight instance. I strongly recommend activating auto-completion in LINQPad -- requires the purchase of an activation code but well worth the price!

After launching LINQPad, you'll first need to register the StreamInsight driver:

Click "Add connection"
Click "View more drivers..."
Click "Download & Enable Driver" under the "Microsoft StreamInsight Driver" heading

To create a connection to the embedded StreamInsight instance, click "Add connection" and select the "Microsoft StreamInsight" context option. For now, just select the "Default Context" option from the "Context Kind" list. Drag the newly created "StreamInsight: Default Context" connection to the query pane or select it from the "Database" list to write a query against a blank-slate context with no built-in streams. You can introduce a data source inline as in the following example that uses the "C# Statement(s)" language option in LINQPad:

var source = new[]
{
    PointEvent.CreateInsert(new DateTime(2011, 1, 1, 12, 0, 0),
                            new { ID = "Hello" }),
    PointEvent.CreateInsert(new DateTime(2011, 1, 1, 13, 0, 0),
                            new { ID = "World" }),
};

var input = source.ToStream(Application,
                            AdvanceTimeSettings.IncreasingStartTime);

input.Dump();

Hit F5 or click "Play" to run!

In addition to the default context, "Performance Counter" and "Hitchhiker'sGuide" contexts are available. The former context allows you to author queries against live performance counter streams. The latter exposes the data explored in A Hitchhiker's Guide to StreamInsight Queries. In addition, we have created a small set of samples -- this set should expand over the next few months! -- that rely on built-in contexts described above. You can add these samples by clicking "Download more samples..." under the "Samples" tab in LINQPad and choosing “Microsoft StreamInsight Samples”. The samples cover:

Queries from the Hitchhiker's Guide
Some simple performance counter queries to get you started
An introduction example using the default context that illustrates various options for displaying StreamInsight query results

In a future blog post, we'll describe how you can author custom StreamInsight query contexts. As always, let us know if you have any questions or comments!

Regards,
The StreamInsight Team

Windows in StreamInsight: The Basics

Roman Schindlauer — Tue, 23 Nov 2010 17:05:00 GMT

If you start learning how to write StreamInsight queries, most likely one of the first concepts you come across are windows. In this article, we’d like to shed some light on the usage of windows, hoping to provide a better understanding for the benefits as well as constraints of using windows. And I will spare you any funny or half-funny puns regarding Microsoft Windows. On a side note, isn’t it striking how people always use the phrase “no pun intended” very intentionally?

In the event processing world, windows are used to define sets, which are then input to some set-based operation. Such operations are either

aggregations, like sum, count, min, max, or a user-defined aggregate, computing a single result for the input set, or
window-based operators, which can return more than one result event for an input set, like TopK or a user-defined operator.

Think of a StreamInsight window as a means to chop up the timeline into finite pieces, each including a number of events. The length of the window can be a fixed time span (hopping & tumbling windows), or determined by the events themselves (count & snapshot windows). Here is an example of a “tumbling window”, with the aggregate “sum” applied to one of the payload fields:

It is important to understand that the window and the set-based operations always need to be used together. The window with the set-based operation can be regarded as a single building block in a StreamInsight query. This is also reflected in the way these constructs are written in LINQ:

var result = from window in input
                 .TumblingWindow(TimeSpan.FromMinutes(5),
                                 HoppingWindowOutputPolicy.ClipToWindowEnd)
             select new { sum = window.Sum(e => e.Value) };

This component – the window together with the set-based operation on top of it – receives a CepStream and produces a CepStream. The window extension method returns the type CepWindowStream, but there is nothing else to do with this type than to apply the set-based operation. We will discuss the specific window types in another posting.

People sometimes ask about different ways to “access” this window, since it must be “just a set of events in memory”, which they want to process somehow. Yes, it is a set of events in StreamInsight-controlled memory, and it actually can be accessed programmatically: in the form of user-defined aggregates (UDAs) or operators (UDOs). These extensibility interfaces let you write C# code against the events contained in a window and produce a single scalar aggregation result (in the case of a UDA) or one or more entire events (with a UDO). Both are implemented as subclasses of types provided by the StreamInsight API. The nice thing about these interfaces is that, even though they introduce imperative code into the StreamInsight event processing flow, the APIs encourage a declarative model: I receive a set of events (the window) and I produce some output for it. I don’t have to worry about when or in what order the UDA or UDO will be called at runtime. And since I receive the set of events as an IEnumerable in this interface, I can write good old LINQ to Objects against them!

Although it might be tempting for developers who haven’t become really comfortable with high-level, declarative programming paradigms such as LINQ to take the “back door” into a UDA/UDO, I would advise to first try expressing a query through built-in StreamInsight operators alone and harness the power of the StreamInsight runtime.

Regards,
Roman