Joining streams based on time #101
Comments
|
Hi @alexmg, the easiest way to do this is to just to create a Union of all three streams, then aggregate over the result using Sum (or another aggregate operator). Also note that you likely will want to Multicast when performing the self-Clip operations. The Union operators will merge the two input streams in order, which means it will not output an event until both streams have caught up to that time. |
|
@peterfreiling 's solution is exactly right, but I want to make sure it's clear why this is the case. You'll notice in Peter's solution that other than the ClipEventDuration operations there are no mentions of any time management or manipulation. It's just a simple union between the streams. Why is this? Internal to Trill, all data is already temporal. Union handles making sure the data is in order, but it also leaves all of the intervals in place. The aggregate Max, like every operation in Trill, is inherently temporal, so you don't need to do anything extra to make it work the way you want it to. @alexmg , it should also be noted that the |
|
Thank you @peterfreiling and @cybertyche for the information. This has helped a lot. I updated my sample to use async Task Main()
{
var start = new DateTime(2019, 06, 26);
var source1 = new[]
{
new Event(start.AddSeconds(1), 1),
new Event(start.AddSeconds(3), 2),
new Event(start.AddSeconds(6), 3),
new Event(start.AddSeconds(8), 2)
}.ToObservable();
var source2 = new[]
{
new Event(start.AddSeconds(2), 2),
new Event(start.AddSeconds(4), 1),
new Event(start.AddSeconds(6), 2),
new Event(start.AddSeconds(8), 1)
}.ToObservable();
var source3 = new[]
{
new Event(start.AddSeconds(2), 3),
new Event(start.AddSeconds(5), 2),
new Event(start.AddSeconds(7), 1)
}.ToObservable();
var stream1 = source1.ToTemporalStreamable(ce => ce.Timestamp.Ticks);
var stream2 = source2.ToTemporalStreamable(ce => ce.Timestamp.Ticks);
var stream3 = source3.ToTemporalStreamable(ce => ce.Timestamp.Ticks);
var query = stream1.Multicast(s1 => s1.ClipEventDuration(s1))
.Union(stream2.Multicast(s2 => s2.ClipEventDuration(s2)))
.Union(stream3.Multicast(s3 => s3.ClipEventDuration(s3)))
.Sum(i => i.Value);
await Display(query);
}
private class Event
{
public DateTime Timestamp { get; set; }
public int Value { get; set; }
public Event(DateTime timestamp, int value)
{
Timestamp = timestamp;
Value = value;
}
public override string ToString()
{
return $"Value: {Value}";
}
}
private static async Task Display<T>(IStreamable<Empty, T> stream, Func<StreamEvent<T>, bool> filter = null)
{
filter = filter ?? (e => true);
await stream.ToStreamEventObservable().ForEachAsync(async r =>
{
if (!filter(r)) return;
switch (r.Kind)
{
case StreamEventKind.Interval:
await Console.Out.WriteLineAsync($"INTERVAL: start={new DateTime(r.StartTime)}, end={new DateTime(r.EndTime)}, payload={r.Payload}");
break;
case StreamEventKind.Start:
await Console.Out.WriteLineAsync($"START EDGE: start={new DateTime(r.StartTime)}, payload={r.Payload}");
break;
case StreamEventKind.End:
await Console.Out.WriteLineAsync($"END EDGE: " +
$"end={new DateTime(r.EndTime)}, original start={new DateTime(r.StartTime)}, payload={r.Payload}");
break;
case StreamEventKind.Punctuation:
await Console.Out.WriteLineAsync($"PUNCTUATION: start={new DateTime(r.StartTime)}");
break;
}
});
}That results in the output below. A value is calculated for I would also like to execute a custom function over the values. Is there a way to get an array of values for each time interval such as @cybertyche What did you have in mind for a contribution around the above windowing pattern? This may be a good way for me to become more familiar with Trill. |
|
To eliminate that first interval from the results of the Union version of the query, and instead wait for all streams to have matched input, you should be able to use Join instead of Union: private static int CustomAggregationFunction(int value1, int value2, int value3) => value1 + value2 + value3;
...
var query =
stream1.Multicast(s1 => s1.ClipEventDuration(s1))
.Join(
stream2.Multicast(s2 => s2.ClipEventDuration(s2)),
(left, right) => new { Item1 = left, Item2 = right })
.Join(
stream3.Multicast(s3 => s3.ClipEventDuration(s3)),
(left, right) => new { left.Item1, left.Item2, Item3 = right })
.Select(tuple => CustomAggregationFunction(tuple.Item1, tuple.Item2, tuple.Item3));Or, if your custom aggregation allows, you can calculate incremental results, like the following for Sum: var query =
stream1.Multicast(s1 => s1.ClipEventDuration(s1))
.Join(
stream2.Multicast(s2 => s2.ClipEventDuration(s2)),
(left, right) => left + right)
.Join(
stream3.Multicast(s3 => s3.ClipEventDuration(s3)),
(left, right) => left + right); |
|
Peter literally just typed exactly what I was in the process of typing. :-) |
|
Of particular note may be this blog entry: https://cloudblogs.microsoft.com/opensource/2019/05/01/trill-102-temporal-joins/ |
|
@alexmg Regarding a contribution, what I'm thinking is something along the lines of this checkin: Another user was interested in having a FullOuterJoin operator (not just a Join and a LeftOuterJoin), so we walked through what the LeftOuterJoin method looks like (it calls down to other methods) and then the user did a PR with the new method. It seems clear that Trill could use a method in the public API that:
Such a method could be implemented just like the FullOuterJoin method, where one does not need to have deep knowledge of the Trill implementation but still needs to understand the algebra to do it right. For instance:
And possibly most importantly, the new method would need a good name. :-) If you're interested, I can start a work item just for this and we could work through the various issues as they come up. |
I am hoping to use Trill for some stream processing scenarios that involve joining multiple streams together based purely on time. The streams being joined are named but the events in the stream do not have identifiers that can be used as a join key. A custom operator should be invoked for each set of events when the end times of all dependent streams are known to have advanced to a point where no more data is expected.
In the example below the custom "sum" operator is invoked with values from the 3 input streams and the result is written into an output stream. The result for the last set of values in the first and second streams is waiting for the end time of the third stream to progress before the result is calculated and written to the output stream.
When a new value is written it replaces any previous value, so the start time of that event becomes the end time of any previous event. I am able to achieve this part by using
ClipEventDurationmethod and passing the source stream itself in as the clip stream.I am not sure how to join the above streams (
query1,query2andquery3) in a way that only considers time and not keys on the individual events.Is it possible to achieve the above scenario with Trill? I am likely missing something obvious due to my lack of familiarity with the API and internals.
The text was updated successfully, but these errors were encountered: