101’ish LINQ Samples for StreamInsight (part 1 - filtering and aggregation)
Following in the great tradition of other learn through example LINQ example pages, such as 101 LINQ Samples, here’s the first in a series of posts on StreamInsight query examples. This will first show up on this blog, then get migrated over to a centralized site on the StreamInsight developer center. We’ve got a fairly hefty queue we’re working through, but if there are specific examples or query patterns of interest, please post them in the comments.
Each query pattern will consist of a description of the input set (.csv file samples for most of these, for ease of use), the full query and the output result. For more complicated query structures the query pattern will also be described visually, using a Visio stencil (to be described in future posts).
The full application framework that I use to run all of these queries, associate with the data sources, etc, is available here. Illustration of the techniques in this codebase will be left for a future post.
In this post:
- Filtering
- Windows and Aggregates
- How can I calculate aggregates, such as averages, over a set of events?
- How can I calculate the average of all events in the last 5 seconds?
- How can I calculate the average of all events in the past 5 seconds, every 2 seconds?
- How can I calculate the average of all events in the past 5 seconds, every time a new event arrives?
- How can I calculate the average of the past 5 events?
- How do I measure the rate of arriving events every two seconds?
- How can I calculate the average of each group of events in the last 10 seconds?
- How can I calculate the average of each group of events in the last 20 seconds, every 10 seconds?
- How can I calculate the average of each group of events in the past 10 seconds, every time a new event in that group arrives?
- How can I find the events with the largest value every 10 seconds?
- How can I find the events with the two smallest values in the last 10 seconds, every 5 seconds?
- How can I find the group of events with the highest average value in the last 10 seconds?
- How can I find the group of events with the highest maximum value in the last 10 seconds, every 5 seconds?
- How can I detect if two events with the same ID value are received within a specific time frame, such as a 5 minute window?
[Where] - How do I filter a stream, keeping only specific events?
var filterQuery = from e in inputStream
where e.Value > 20
select e;
[Input] Result:
|
How do I filter on multiple conditions?
var filterQuery = from e in inputStream
where e.Value > 70 && e.SensorId == 1001
select e;
[Input] Result:
|
How can I calculate aggregates, such as averages, over a set of events?
All aggregates, such as averages, min, max, etc are calculated over a set of events. Sets of events in StreamInsight are defined as those events falling into a window. Windows operate over a period of time (hopping, tumbling and sliding), or over a specific number of events (count).
How can I calculate the average of all events in the last 5 seconds?
var query = from window in inputStream.TumblingWindow(
TimeSpan.FromSeconds(5),
HoppingWindowOutputPolicy.ClipToWindowEnd)
select new
{
Average = window.Avg(e => e.Value)
};
[Input] Result:
|
How can I calculate the average of all events in the past 5 seconds, every 2 seconds?
var query = from window in inputStream.HoppingWindow(
TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(2),
HoppingWindowOutputPolicy.ClipToWindowEnd)
select new
{
Average = window.Avg(e => e.Value)
};
[Input] Result:
|
How can I calculate the average of all events in the past 5 seconds, every time a new event arrives?
var query = from window in inputStream
.AlterEventDuration(e => TimeSpan.FromSeconds(5))
.SnapshotWindow(SnapshotWindowOutputPolicy.Clip)
select new
{
Average = window.Avg(e => e.Value)
};
[Input] Result:
|
How can I calculate the average of the past 5 events?
Aggregates over a set number of events are produced using Count windows. As of StreamInsight v1, Count Windows cannot use the built-in aggregate functions (Count, Avg, Min, Max), but must instead use user-defined aggregates. The code sample below shows the use of a simple averaging function with a count window with the user defined aggregate.
var query = from window in inputStream.CountByStartTimeWindow(
5, CountWindowOutputPolicy.PointAlignToWindowEnd)
select new
{
Average = window.UserDefinedAverage(e => e.Value)
};
public class SimpleAggregate : CepAggregate<double, double>
{
public override double GenerateOutput(IEnumerable<double> payloads)
{
return payloads.Average();
}
}
public static class MyExtensions
{
[CepUserDefinedAggregate(typeof(SimpleAggregate))]
public static double UserDefinedAverage<InputT>(this CepWindow<InputT> window,
Expression<Func<InputT, double>> map)
{
throw CepUtility.DoNotCall();
}
}
[Input] Result:
|
How do I measure the rate of arriving events every two seconds?
var query = from window in inputStream.TumblingWindow(
TimeSpan.FromSeconds(2),
HoppingWindowOutputPolicy.ClipToWindowEnd)
select new
{
Count = window.Count()
};
[Input] Result:
|
How can I calculate the average of each group of events in the last 10 seconds?
For this example, we’ll calculate the average of each sensor every 10 seconds. We do this by first grouping the events by the sensor ID, then performing the same aggregation operation from the average without grouping (with the addition of the group key – in this case, the sensor ID).
var query = from e in inputStream
group e by e.SensorId into sensorGroups
from window in sensorGroups.TumblingWindow(
TimeSpan.FromSeconds(10),
HoppingWindowOutputPolicy.ClipToWindowEnd)
select new
{
SensorId = sensorGroups.Key,
Average = window.Avg(e => e.Value)
};
[Input] Result:
|
How can I calculate the average of each group of events in the last 20 seconds, every 10 seconds?
var query = from e in inputStream
group e by e.SensorId into sensorGroups
from window in sensorGroups.HoppingWindow(
TimeSpan.FromSeconds(20), TimeSpan.FromSeconds(10),
HoppingWindowOutputPolicy.ClipToWindowEnd)
select new
{
SensorId = sensorGroups.Key,
Average = window.Avg(e => e.Value)
};
[Input] Result:
|
How can I calculate the average of each group of events in the past 10 seconds, every time a new event in that group arrives?
var query = from e in inputStream
.AlterEventDuration(e => TimeSpan.FromSeconds(10))
group e by e.SensorId into sensorGroups
from window in sensorGroups
.SnapshotWindow(SnapshotWindowOutputPolicy.Clip)
select new
{
SensorId = sensorGroups.Key,
Average = window.Avg(e => e.Value)
};
[Input] Result:
|
How can I find the events with the largest value every 10 seconds?
var query = (from window in inputStream.TumblingWindow(
TimeSpan.FromSeconds(10),
HoppingWindowOutputPolicy.ClipToWindowEnd)
from e in window
orderby e.Value descending
select e).Take(1);
[Input] Result:
|
Note that for the first group of events, the are three events with the same maximum value (80) in the same window of time. Thusly, all of the events with that maximum value are returned.
How can I find the events with the two smallest values in the last 10 seconds, every 5 seconds?
var query = (from window in inputStream.HoppingWindow(
TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(5),
HoppingWindowOutputPolicy.ClipToWindowEnd)
from e in window
orderby e.Value ascending
select e).Take(2);
[Input] Result:
|
Note the same effect herein – multiple events having identical minimum values in the window.
How can I find the group of events with the highest average value in the last 10 seconds?
This query is performed in two stages. In the first query we determine the average value of each sensor over the past 10 seconds (using a group by, a tumbling window and an Avg()). Then we take the output of that aggregate and perform a snapshot on top of that (against each set of events from the first query), sort it and take the top value.
var avg = from e in inputStream
group e by e.SensorId into sensorGroups
from window in sensorGroups.TumblingWindow(
TimeSpan.FromSeconds(10),
HoppingWindowOutputPolicy.ClipToWindowEnd)
select new
{
SensorId = sensorGroups.Key,
Average = window.Avg(e => e.Value)
};
var query = (from win in avg
.SnapshotWindow(SnapshotWindowOutputPolicy.Clip)
from e in win
orderby e.Average descending
select e).Take(1);
[Input] Results:
|
How can I find the group of events with the highest maximum value in the last 10 seconds, every 5 seconds?
var avg = from e in inputStream
group e by e.SensorId into sensorGroups
from window in sensorGroups.HoppingWindow(
TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(5),
HoppingWindowOutputPolicy.ClipToWindowEnd)
select new
{
SensorId = sensorGroups.Key,
Maximum = window.Max(e => e.Value),
Minimum = window.Min(e => e.Value)
};
var query = (from win in avg
.SnapshotWindow(SnapshotWindowOutputPolicy.Clip)
from e in win
orderby e.Maximum descending
select e).Take(1);
[Input] Result:
|
Again, note the StreamInsight engine delivering the values corresponding to the top maximum value (80 or 81).
How can I detect if two events with the same ID value are received within a specific time frame, such as a 5 minute window?
To implement this query, we use a sliding window to look at arriving events in a 5 minute window, grouped by their ID value. We then count the number of events in the window with that ID, and filter out groups that have less than 2 events. Or, to say it another say, we only want groups (sensors) that have more than 1 event, thus removing groups with only 1 event (there are no groups with zero :).
var eventCount = from e in inputStream.AlterEventDuration(e => TimeSpan.FromMinutes(5))
group e by e.SensorId into sensorGroups
from win in sensorGroups.SnapshotWindow(SnapshotWindowOutputPolicy.Clip)
select new
{
SensorId = sensorGroups.Key,
Count = win.Count()
};
var query = from e in eventCount
where e.Count >= 2
select e;
[Input] Result:
|