microsoft / trill Goto Github PK
View Code? Open in Web Editor NEWTrill is a single-node query processor for temporal or streaming data.
License: MIT License
Trill is a single-node query processor for temporal or streaming data.
License: MIT License
[Fact]
public void PartitionedStreamEvent_EndStart_Stitching_Test()
{
PartitionedStreamEvent<string, IONRecord> last;
var input = new List<PartitionedStreamEvent<string, IONRecord>>
{
(last = PartitionedStreamEvent.CreateStart(key: "c1", startTime: 1, payload: new IONRecord { Id = "c1", Data = "shortdes1" })),
PartitionedStreamEvent.CreateEnd(key: "c1", endTime: 7, originalStartTime: last.StartTime, payload: last.Payload),
(last = PartitionedStreamEvent.CreateStart(key: "c1", startTime: 7, payload: new IONRecord { Id = "c1", Data = "shortdes1" }))
}.ToObservable().ToStreamable();
UnitTests_Funcs.RunAndAssert(new[]
{
" # 01234567 ",
" 1 (c1)[Start: 1, {Id:c1, Data:shortdes1}] [ ",
" 2 (c1)[End: 7,1, {Id:c1, Data:shortdes1}] ------) ",
" 3 (c1)[Start: 7, {Id:c1, Data:shortdes1}] [ ",
" 4 [LowWatermark: +inf] ",
}, input, onlyData: false);
var output = input.Stitch();
UnitTests_Funcs.RunAndAssert(new[]
{
" # 01234567 ",
" 1 (c1)[Start: 1, {Id:c1, Data:shortdes1}] [ ",
" 2 (c1)[End: 7,1, {Id:c1, Data:shortdes1}] ------) ",
" 3 (c1)[Start: 7, {Id:c1, Data:shortdes1}] [ ",
" 4 [LowWatermark: +inf] ",
}, output, onlyData: false);
}
my input data is already partitioned by a key and it would be an overhead to ingress it all merged only to repartition it later with groupjoin
thanks
In the following code I am saving a Streamable to disk, then restoring it such that I can execute queries on it. The first one is fine, but I get an exception with the second.
All I am doing here is issuing a Count query twice. Doing some digging, I found that the position on the Stream is at the end after the first query. If you uncomment the Position manipulation, the second query succeeds.
Is the behaviour I am seeing intended? I feel like this is going to make Streambles backed by on-disk Streams very difficult to use.
internal struct Payload
{
public long field1;
public long field2;
}
public static void Test()
{
Random random = new Random(Guid.NewGuid().GetHashCode());
string filePath = @"C:\temp\streamtest.bin";
var dataset = Observable.Range(0, TotalInputEvents)
.Select(e => StreamEvent.CreatePoint(0, new Payload { field1 = e, field2 = random.Next(100, 250) }))
.ToStreamable();
using (var stream = File.Create(filePath))
{
dataset.ToBinaryStream(stream, writePropertiesToStream: true);
}
using (var stream = File.OpenRead(filePath))
{
var streamable = stream.ToStreamable<Payload>(readPropertiesFromStream: true);
// long position = stream.Position;
streamable.ToStreamEventObservable().Where(e => e.IsPoint).Count().ForEachAsync(e => Console.WriteLine(e)).Wait();
// stream.Seek(position, SeekOrigin.Begin);
streamable.ToStreamEventObservable().Where(e => e.IsPoint).Count().ForEachAsync(e => Console.WriteLine(e)).Wait();
}
}
Unhandled Exception: System.AggregateException: One or more errors occurred. (Invalid integer long in the input stream.) ---> System.Runtime.Serialization.SerializationException: Invalid integer long in the input stream.
at Microsoft.StreamProcessing.Serializer.BinaryDecoder.DecodeLong() in D:\a\1\s\Sources\Core\Microsoft.StreamProcessing\Serializer\Encoders\BinaryDecoder.cs:line 88
at lambda_method(Closure , BinaryDecoder )
at Microsoft.StreamProcessing.BinaryIngressReader`2.Ingress(IStreamObserver`2 observer) in D:\a\1\s\Sources\Core\Microsoft.StreamProcessing\Ingress\Binary\BinaryIngressReader.cs:line 68
--- End of inner exception stack trace ---
at System.Threading.Tasks.Task.Wait(Int32 millisecondsTimeout, CancellationToken cancellationToken)
at System.Threading.Tasks.Task.Wait()
at PerformanceTest.Program.Test() in C:\Git\TrillSamples\TrillSamples\PerformanceTest\Program.cs:line 87
at PerformanceTest.Program.Main(String[] args) in C:\Git\TrillSamples\TrillSamples\PerformanceTest\Program.cs:line 148
The following statement breaks tests on non-US machines:
https://github.com/Microsoft/Trill/blob/3ca9a12ec8386c1d58e6bd32a665ba41a6909769/Sources/Core/Microsoft.StreamProcessing/StringHandling/MultiString.cs#L108
Maybe it would help to configure test runner to use hardcoded culture if its unavoidable, and mention it in documentation?
Thank you.
Bug in joins?
public struct IONRecord
{
public string Id;
public string Data;
public override string ToString() => $" {{{nameof(Id)}:{Id}, {nameof(Data)}:{Data}}}";
}
[Fact]
public void LeftOuterJoin_FlushProblem_Test()
{
var left_subject = new Subject<StreamEvent<IONRecord>>();
var right_subject = new Subject<StreamEvent<IONRecord>>();
var left_stream = left_subject.ToStreamable(disorderPolicy: DisorderPolicy.Throw(), flushPolicy: FlushPolicy.FlushOnPunctuation, periodicPunctuationPolicy: null);
var right_stream = right_subject.ToStreamable(disorderPolicy: DisorderPolicy.Throw(), flushPolicy: FlushPolicy.FlushOnPunctuation, periodicPunctuationPolicy: null);
var output_events = new List<string>();
var output_stream = left_stream
.LeftOuterJoin(right: right_stream,
leftKeySelector: l => l.Id,
rightKeySelector: r => r.Id,
outerResultSelector: l => new
{
l_Id = l.Id,
r_Id = (string)null
},
innerResultSelector: (l, r) => new
{
l_Id = l.Id,
r_Id = r.Id
});
var output_observable = output_stream.ToStreamEventObservable(reshapingPolicy: ReshapingPolicy.None);
output_observable.Subscribe(onNext: se =>
{
if (se.IsData) output_events.Add(se.ToString());
});
//Start with no output events
Assert.Equal(0, output_events.Count);
//No flush on first left is ok (I guess...)... waiting for sync time to move forward
left_subject.OnNext(StreamEvent.CreateStart(701, new IONRecord { Id = "c1", Data = "shortdes1" }));
Assert.Equal(0, output_events.Count);
//Eh? Sync time on the left moved forward... but no flush yet
left_subject.OnNext(StreamEvent.CreatePunctuation<IONRecord>(punctuationTime: 702));
Assert.Equal(0, output_events.Count);
//Only flushing when both, left and right move forward... i thought in joins the sync time is supposed
//to be the max sync time from the left and right
right_subject.OnNext(StreamEvent.CreatePunctuation<IONRecord>(punctuationTime: 702));
Assert.Equal(new[] {
"[Start: 701,{ l_Id = c1, r_Id = }]",
}, output_events);
}
At present, columnar data batches are only supported for non-polymorphic data types. They do not support interfaces, nor do they support non-sealed classes.
Consider adding support for Apache Beam's unified model for defining both batch and streaming data-parallel processing pipelines.
All ingress methods into Trill currently rely on IObservable and thus rely on push semantics. If one wants to ingress data from an IEnumerable, one must first use an external framework or custom code to convert the IEnumerable to an IObservable. For multi-input queries, doing so caused potential threading issues.
We would like to add first-class support for IEnumerable ingress. Doing so will likely require implementation of back pressure mechanisms in binary operators, where data is drawn from one side or the other to maintain optimal data flow (instead of relying on the ingress operators to handle control flow).
There is Join(...) and LeftOuterJoin(...)
How about FullJoin ?
Support for Apache Arrow which is a cross-language development platform for in-memory data. It specifies a standardized language-independent columnar memory format for flat and hierarchical data, organized for efficient analytic operations on modern hardware. It also provides computational libraries and zero-copy streaming messaging and interprocess communication.
Apache Arrow is backed by key developers of 13 major open source projects, including Calcite, Cassandra, Drill, Hadoop, HBase, Ibis, Impala, Kudu, Pandas, Parquet, Phoenix, Spark, and Storm making it the de-facto standard for columnar in-memory analytics.
Related issues:
One can always create custom adapters for data in any format and convert it into IObservable or IObservable<StreamEvent>. However, for some operations, it may instead make sense to instead compile the query into operators that operate directly over data in its native format. For instance, one may instead of having a batch of T payloads or a batch of columns of T, one may have a batch of data in parquet format.
The current heuristic for data flowing through Trill is that if it begins in "row mode" it will stay in row mode for the duration of the query. If it is in "column mode" it will stay that way until it hits an operator that cannot operate on columnar data, so it switches to "row mode". We may be able to allow an extensibility point wherein we allow a query to operate over data in "parquet mode" or "JSON mode" or some other mode, and then when it can no longer do so, switch to either "column mode" or "row mode".
Hello, thank you very much for designing and open sourcing this system. I've been reading y'alls papers on Trill, FASTER and Quill for years now.
I have a few questions about how to use it.
I realize that Trill is primarily a streaming data engine. But I work with a lot of batched data also that i would like to perform queries against using Trill.
Maybe i don't have the egress side of things setup correctly but my setup looks like this
public static async Task EventsPerHour(string rootFolder)
{
await
LogExtractor
.Create()
.ExtractSingleSiteDirectory(rootFolder)
.AsObservable()
.ToTemporalStreamable(
e => e.Entry.Timestamp.Ticks,
DisorderPolicy.Drop(TimeSpan.TicksPerHour),
FlushPolicy.None,
PeriodicPunctuationPolicy.None())
.GroupApply(
e => e.Entry.Timestamp.Hour,
g => g.Count(),
(g, v) => new { Hour = g.Key, Count = v })
.ToStreamEventObservable()
.ForEachAsync(i => {
Console.WriteLine(i.Payload);
});
}
So what i would like to get out of this somehow is a single list of (Hour, Count) pairs.
What i actually get is a lot of incremental updates as the data flows in. To compensate i made a handler method that tracks all the updates per group and only keeps the last one. It produces correct output, but it seems wasteful to have the engine continue to produce output that i'm discarding.
Can i tell Trill to withhold output until the input stream terminates? If so how?
Also, why are so many things marked [EditorBrowsable(EditorBrowsableState.Never)]
? For example, I see people using the 3 argument version of group apply in examples, but for whatever reason you have GroupSelectorInput<T>.Key
marked as never browseable, making the result selector function seem useless initially.
Is there a reason this property (and others like it) is hidden?
Are these two constructions equivalent? If so which should i prefer?
.GroupApply(
e => e.Entry.Timestamp.Hour,
g => g.Count(),
(g, v) => new { Hour = g.Key, Count = v })
//yield type `IStreamable<Empty,'a>
vs.
.Partition(e => e.Entry.Timestamp.Hour)
.Aggregate(g => g.Count())
.SelectByKey((time, key, count) => new { Hour = key, Count = count })
//yields type IStreamable<PartitionKey<int>, 'a>
In the current implementation, generated code artifacts are created on demand. That implies a code generation session for each generated operator, a generation session for each generated batch subtype, and a generation session for each generated dictionary subclass.
We speculate that if we can gather all code generation needs into a single code generation session to occur just before data source subscription that startup time for the query should decrease.
Note: If this feature does not result in a significant gain in startup performance, or it causes runtime throughput or latency regression, it should be abandoned.
I run into an exception when trying to dispose a subscription.
For example:
subscription = observableInput.ToAtemporalStreamable(TimelinePolicy.Sequence(1)).Count().ToStreamEventObservable().Where(c => c.IsData).Subscribe
My input feed will continue to deliver input data, but if I dispose of the subscription then this will result in a null reference exception.
I did some analysis on the codebase and noticed the following:
DisposeState method in SubscriptionBase will call the disposer?.Dispose() method. However the "disposer" field is never set.
In addition it releases the "currentBatch" buffer before any dispose call as well.
The result is that the currentBatch is set to null before the dispose call. This means another OnNext could happen and this will then result in a null reference on the currentBatch.
A possible solution is to make sure the "disposer" field is set to the subscription and the diposer?.Dispose() should be called first in the DisposeState method in order to properly dispose of the observable chain.
There is a second problem with the MonotonicSubscriptionWallClock. This will override the DisposeState method and never calls the base implementation, which means it never disposes of the subscription and keeps on processing events after the subscription has been disposed of.
When would someone use that method vs say managing the Flushes with say Punctuations ?
At the moment, operator fusion of select, select many, where, and alter duration by constant operators is in place for ingress. Those operations can effectively become part of the ingress operation, localizing some of the computation.
We would like to do the same for join operations, allowing that subset of operations to fuse into the output of a join operator. The result will:
Consider adding support for System.Threading.Channels & C# 8 IAsyncEnumerable
Streams are very FP-friendly concept.
But C# is not very FP friendly language, that's why making ETL processes with F# is much easier (currying, pipelining etc).
Do you have any plans on making F# friendly facade?
Cannot run perf test. The error is in a generated file so cannot set a breakpoint and debug it easily as well. The code in the PerformanceTesting project hasn't changed since the initial commit and there is TODO in the generated file, so the question is if the test is supposed to work at all at the moment?
or class?
Implement a progressive version of the quantize operator
I want combine my input streams with some non-stream-based reference data which is big (many records) and slow-moving (it changes, but no where near the speed of my input streams). How do I approach I best approach this with Trill?
I could think of a couple of ways of approaching it, but neither seemed ideal—and may not even be valid anyway (I'm new to Trill)—so I wanted to ask if there were a recommended solution for such a scenario.
I have a input stream of sensor data and a database which stores a relationship between a particular sensor and the operator who owns it.
I take my sensor data input stream, apply some fancy heuristics and determine outliers. I want to notify the operator who owns the sensor of those outliers, so I need to join these events with that database table somehow.
Lift the rows from my database table to a streamable and use a Trill join operator.
var owners = db.Query<Owner>("select sensor_id, owner_email_address from sensor_owner")
.ToObservable()
.ToStreamable()
.Cache()
sensorStream
.DoFancyHeuristics()
.Join(owners, l => l.SensorId, r => r.SensorId, (l, r) => new {
l.SensorId,
l.Measurement,
r.OwnerEmailAddress
})
.SendEmailEtc()
However, what if that sensor_owner table is freaking large? (Or we don't have any viable way of scraping the entire set of relationships?)
What about when there are changes made to it? (We could refresh our streamable cache, but we might not be able to do anything smarter than refreshing all of the rows.)
We could instead defer a (specific) relationship query till after there's one of these outliers events, which brings me to...
Leave the realm of streamables and use an Rx operator.
sensorStream
.DoFancyHeuristics()
.ToStreamEventObservable()
.Select(e => db
.Query<Owner>("select sensor_id, owner_email_address from sensor_owner where sensor_id = $1", e.SensorId)
.Select(owner => new { e.SensorId, e.Measurement, owner.OwnerEmailAddress })
.ToObservable())
.Concat()
.ToStreamable()
.SendEmailEtc()
Here we're paying the cost of a network call on each outlier because we don't want to (or can't) pay the cost of storing it all in memory.
Leaving and re-entering the realm of streamables seems a little off intuitively, would it actually be fine in practice? One thing that comes to mind is if I'm using the HA feature, now I have two checkpoints to take so I no longer have a consistent snapshot of my query state.
(I understand in this scenario I could very well not return to streamables, but in similar scenarios I might want to continue composing operators.)
I understand I could leave streamables after I've translated my sensor data to my outlier events, but the complexity of the query is likely to evolve in ways which would very much like me to remain up in the realm of streamables. (e.g. now I want to batch these outliers into one notification to operators.)
I think I could achieve this succinctly if there were support for higher-order streamables in future, but I also understand that such a thing might not be appropriate or compatible with the design of Trill. (e.g. something like Rx's Merge operator not as useful when you add the strict temporal semantics of Trill, though still possible I think...)
sensorStream
.DoFancyHeuristics()
.Select(e => db
.Query<Owner>("select sensor_id, owner_email_address from sensor_owner where sensor_id = $1", e.SensorId)
.Select(owner => new { e.SensorId, e.Measurement, owner.OwnerEmailAddress })
.ToObservable()
.ToStreamable())
.Concat()
.SendEmailEtc()
Currently, the Trill ingress code only allows a single disorder policy to be specified. Research has been done to allow that restriction to be loosened to multiple disorder policies set at different lag allowances. However, this work has not yet been imported into the main code base because of API issues - we need to figure out how to integrate the work more seamlessly.
Research paper link:
https://ieeexplore.ieee.org/document/8509288
The present (default) implementation of Trill has all memory management done by .Net and in-memory. We are in the process of making implementations of Trill operators that use the FASTER key-value store engine for state management instead. Doing so will allow the option of having operator state spanning both memory and disk, and would allow for potentially faster checkpointing and recovery.
Should the FASTER version of operators prove to have similar or better performance than their previous versions, we will remove the previous versions in favor of the new ones to reduce code duplication.
Repo link for FASTER:
https://github.com/microsoft/faster
Implement progressive versions of hopping and tumbling windows:
.Net 4..7 introduced the Span data type for better direct memory access. This feature, as well as potentially many others, could be used to either simplify the Trill code base or to improve performance for users of later framework versions.
Hello,
I have the following query, my issue is that, I don't get any output on my first publication to
parentSubject and childSubject, but I do start to receive output on the subsequent publications, but form previous publications. For example, I publish A, but I don't get result of A, until I publish B.
My impression was that, streams would get flushed every 1 and 2 seconds and produce output on every flush.
I create events using StreamEvent.CreateStart
`
var parentStreamable = parentSubject
.ToStreamable(null, FlushPolicy.FlushOnPunctuation, PeriodicPunctuationPolicy.Time((ulong)TimeSpan.FromSeconds(1).Ticks));
var clippedParentStreamable = parentStreamable
.ClipEventDuration(parentStreamable, x => x.Id, x => x.Id);
var childStreamable = childSubject
.ToStreamable(null, FlushPolicy.FlushOnPunctuation, PeriodicPunctuationPolicy.Time((ulong)TimeSpan.FromSeconds(2).Ticks));
var clippedChildStreamable = childStreamable
.ClipEventDuration(childStreamable, x => x.Id, x => x.Id);
var output = clippedParentStreamable.Where(x => x.State == OrderState.Open)
.Join(clippedChildStreamable.Where(x => x.State == OrderState.Open),
x => x.Id,
x => x.ParentId,
(left, right) => new {Parent = left, Child = right})
.GroupApply(
x => x.Parent,
x => x.Count(),
(key, count) => new {ParentOrder = key.Key, ChildCount = count})
.Where(x => x.ChildCount > 0);`
At the moment, operator fusion of select, select many, where, and alter duration by constant operators is in place for ingress. Those operations can effectively become part of the ingress operation, localizing some of the computation.
We would like to do the same for aggregate operations, allowing that subset of operations to fuse into the output of an aggregate operator. The result will:
Hello,
I'd like to experiment with a Trill
lib a little bit and was wondering what is the best approach for use case where there is a single 'source stream' for example trading data, like individual trades and many queries for such source stream, where first query would be Where
that would narrow down trades to individual instruments. In a writing queries guide I've read about multicast that is necessary for such use case, but when looking at the source code I've also found about Publish
.
I'm not really familiar with rx
so that is a little bit confusing for me when to use Multicast
vs Publish
, what would you suggest? Or perhaps it's better to create separate streams instead of single one for each Where
so then multicast is not necessary? Do you know any guidelines, lessons learned about that? I'd love to read more, but couldn't find anything in docs.
I've also seen mentions about partitioning in the source code, is this only related to group
operator or something that could also be useful for my use case?
I'll be setting low batch size (< 5, maybe less), are there any settings that I could tweak for very near real-time queries to get best perf, sacrificing throughput, but getting lowest latency possible ?
Thanks a lot!
Requirement 1: Create an analogue to IQueryable or IQbservable for IStreamable.
That means having a new interface, IQStreamable, and a provider class, IStreamProvider, that can build up an expression in LINQ and then evaluate it later. The expression is inspectible and manipulable just like other LINQ expressions.
Requirement 2: Ensure that the provider can stand independently of the engine.
Not only would this allow someone to implement an engine to the Trill API with different characteristics, but it would also allow the API implementation to evolve independently of the API if it needs to.
Biggest open question: Do we precisely mirror the existing IStreamable API, or do we instead try to implement API 2.0 with lessons learned?
For the most part, we are incredibly proud of our API. However, there are a few places where we would make changes if we could. Those places are:
i have a stream of the following 6 events
A1 , B1 , B2 , A2 , B3 , B4
and I want to produce another stream of the B's, together with the last A's
(B1,A1) , (B2,A1) , (B3,A2) , (B4,A2)
how can I go about it?
I'd like to do a LEFT OUTER JOIN between two streams
It looks like I could use DefinePattern. Is there any other better way?
Thanks,
Nestor
I’d submit that SyncTime and OtherTime should not be internal but public in the StreamEvent struct.
They are both integral concepts of the library.
For the netstandard2.0 build, the XML documentation is missing in the NuGet package.
The nuspec definitions for the net46 assembly do not seem to be in line with the net standard version.
For example it declares the following dependency:
The net standard has far fewer dependencies listed and refers to 2.9.0.
The result is that I get assembly conflicts and use of the 1.3.2 version instead of the 2.9.0 version.
Could you update the nuspec definition and publish to nuget?
Is there any sample code available that demonstrates how to write to and read from binary streams?
Currently I have the following adapted from one of the existing Trill samples:
var data = CreateStream(false);
string filePath = @"C:\temp\trill_stream.bin";
using (var stream = new FileStream(filePath, FileMode.Open))
{
data.ToBinaryStream(stream, true);
}
var fromFile = filePath.ToStreamableFromFile<SensorReading>();
fromFile.ToStreamEventObservable().ForEachAsync(e => Console.WriteLine(e)).Wait();
When I run this code I get a null reference exception on that last line with the following stack trace:
at Microsoft.StreamProcessing.StreamEventEgressPipe`1.OnNext(StreamMessage`2 batch) in D:\a\1\s\Sources\Core\Microsoft.StreamProcessing\Egress\Temporal\TemporalEgressPipe.cs:line 28
at Microsoft.StreamProcessing.BinaryIngressReader`2.Ingress(IStreamObserver`2 observer) in D:\a\1\s\Sources\Core\Microsoft.StreamProcessing\Ingress\Binary\BinaryIngressReader.cs:line 69
This code below from AfaMultiEventListTransformer.cs
looks strange, but I can't quite tell if it is a bug or just a candidate for some code-cleanup?
Is a different payloadRepresentation
value supposed to be used if Config.ForceRowBasedExecution
== true?
Currently both branches of the C ? A : B
statement below end up making the same assignment with a new ColumnarRepresentation(typeof(TPayload))
If this code should always assign the same new ColumnarRepresentation, then I can easily submit a PR to simplify and clarify the code down to:
var payloadRepresentation = new ColumnarRepresentation(typeof(TPayload));
Performance numbers for FFT in the digital signal processing library were based on a native implementation of the algorithms in .Net Framework. The OSS release of Trill and all of its features must be .Net Standard 2.0 compliant, and it cannot rely on any library for which there is not proper licensing. We need to do one of the following:
I changed my payload from struct to class, and now after doing a join I get the error:
Type of left side of join, 'xxxxxxx', does not have a valid equality operator for columnar mode.
Can you help me fast? (usually it takes many days to get a reply here... it'd be great to either have better documentation, or more prompt responses … thanks much)
See the following code:
[Fact]
public void QueryContainer_MultipleOutputs_Test()
{
var container = new QueryContainer();
var asset_subject = new Subject<StreamEvent<IONRecord_Struct>>();
var assets_input = container.RegisterInput(asset_subject, identifier: "input1");
var prices_subject = new Subject<StreamEvent<IONRecord_Struct>>();
var prices_input = container.RegisterInput(prices_subject, identifier: "input2");
var join = assets_input.Join(
right: prices_input,
leftKeySelector: l => l.Id,
rightKeySelector: r => r.Id,
resultSelector: (l, r) => new
{
l.Id,
l.Data,
r_Data = r.Data,
});
container.RegisterOutput(join, identifier: "output1").Subscribe(onNext: p => Debug.Print($"{p}"));
var left_join = assets_input
.LeftOuterJoin(right: prices_input,
leftKeySelector: l => l.Id,
rightKeySelector: r => r.Id,
outerResultSelector: l => new
{
l.Id,
l.Data,
r_Data = string.Empty,
},
innerResultSelector: (l, r) => new
{
l.Id,
l.Data,
r_Data = r.Data
});
//NEXT LINE THROWS !??
//System.InvalidOperationException: 'Operation is not valid due to the current state of the object.'
container.RegisterOutput(left_join, identifier: "output2").Subscribe(onNext: p => Debug.Print($"{p}"));
var process = container.Restore(inputStream: null);
}
Trying to write code for a stream containing temperature values and if they are above a specified threshold (i.e 70 degrees) for more than 20 minutes, then output a "TEMPERATURE LIMIT EXCEEDED" error value. Once the condition is no longer satisfied, output a "TEMPERATURE NORMAL" value.
Its probably quite straight forward but I currently struggling to work out how to approach this since I need to calculate the passing of time between values if the condition is met.
Does Trill have one notion of time across all streams or only per stream?
If it does, is there a way to access that long property?
Consider open sourcing the Kusto Query Language to Trill logic converter/mapping library.
Thoughts, opinions, comments @cybertyche ?
Hi,
I just want to confirm this is client side only streaming and it can work with any pub/sub systems correct?
The prototype/research implementation of DSP over Trill has been developed, but it is not in product-ready shape. It needs a thorough API review, and a cleanup of the code, as well as tests added.
Key concepts to review:
Research paper:
https://dl.acm.org/citation.cfm?doid=3035918.3035935
For a data structure with fields a, b, and c, if the downstream query operators never refer to field a directly but instead refer to a.d.z, or a["bacon"], or some other constant expression, it may make sense to have a column representing a.d.z or a["bacon"] instead of a. This change would require an alteration of the data type structure of the generated columnar batch, and it would change the way that generated operators over those columns reference fields.
At present, the columnar data format creates columns for every field and property in the type. For instance, a struct with fields a, b, and c will yield batches that have arrays for a, b, and c as well. However, if no downstream operators require access to field b, it makes no sense to allocate any arrays for it let alone populate them. This change need not actually change the type of the generated columnar format, just that the arrays for unneeded fields don't get allocated or populated.
I'm looking at Trill and it seems similar to Rx. In fact Trill seems to use IObservable for ingress and egress to the pipeline. I'm trying to understand the differences. Some that I see are:
Areas where it is the same:
Questions:
Is it different on:
I'll probably ask more but this I figure should be the start of a conversation or at least a FAQ item.
There are already data policies at ingress for data that arrives "late". We can drop, adjust, or throw when data arrives late, and we can hold data in reserve for a certain period of time to allow some reordering.
However, if a data point arrives "too early" we do not have a way to deal with it currently. For instance, if the current data time is X, and the next data point arrives with a timestamp of X + 2 days, this may be a result of:
Today, however, we accept this value as valid and current, and the sync time is advance all the way to X + 2 days. Any further data will now be compared against the new sync time, and thus data may come start to get dropped or adjusted improperly.
We would like to add an ingress policy that allows for a threshold to be specified for maximum sync time advancement. If a data value arrives so far into the future that the maximum advancement is exceeded, the value is either:
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.