问题标签 [streaminsight]

我正在使用基于样本的 WCF Observables：

var observableWcfSource = app.DefineObservable(() => new WcfObservable())

它可以工作，但不会在服务器中注册源，因此当我使用事件流调试器探索我的独立 StreamInsight 服务时，我看不到源（并在其他查询中使用它）。如何注册我的来源？

我一直在使用 StreamInsight v2.3 和它提供的更新的 Rx 功能。我正在研究将 SI 用于事件溯源实施。我已经调整了一些MSDN 示例代码以获得以下内容：

服务器进程的代码：

客户端进程代码：

尝试运行它会产生以下错误：

不支持从远程 'System.Reactive.Linq.IQbservable`1[System.Int64]' 读取。使用“Microsoft.ComplexEventProcessing.Linq.RemoteProvider.Bind”方法通过远程观察者从源中读取。

我开始使用的示例代码定义了一个观察者和接收器，并将其绑定在 StreamInsight 服务器中。我试图让观察者保持在客户端进程中。有没有办法在客户端应用程序中为远程 StreamInsight 源设置观察者？这是否必须通过客户端观察到的服务器中的 WCF 端点之类的东西来完成？

我对启用了 WCF 的源和接收器进行了查询，并使用了嵌入在我的项目中的 StreamInsight。它在我运行查询时创建源。没关系，但在某些数据元组进入源（并使其通过接收器）之前，它不会创建接收器。因此，在输入数据到达我的查询之前，我的客户端无法连接到接收器（通过 WCF）。我怎么解决这个问题？！

如何定期轮询相对静态的源（如数据库）以在 Microsoft StreamInsight 中创建参考流？

这是我尝试过的。我将用户元数据的数据库表示为一个简单的List<UserMetaData>

这是 UserMetaData 类

其余示例代码替换标准 StreamInsight 嵌入式部署设置中的省略号。

首先，我创建一个这样的心跳：

在实际应用程序中，我可能会将此心跳间隔设为五分钟而不是两秒。无论如何，接下来我希望 heatbeat 触发新用户元数据的数据库查找：

IQbservable.SelectMany 扩展应该使IEnumerable<UserMetaData>我期望的 referenceData 变平。该_参数丢弃从心跳发出的长。然后ToPointStreamable将 转换IObservable<UserMetaData>为IQStreamable开始时间为现在的点事件。（DateTime.Now 可能不是很 StreamInsight-y）

然后我将它转换为一个信号，通过一个简单的查询运行它，定义一个控制台接收器并部署它。

我的最后一步是去Bind水槽。我还将等待几秒钟以查看我的元数据轮询心跳打印到屏幕上的输出，然后将新UserMetaData记录添加到我的数据库并等待查看更改是否得到反映。

新的 UserMetaData 记录永远不会反映在输出中

我假设正在发生的是我的 UserMetaData 列表正在被序列化并在 SI 服务器上重新创建，因此对我的本地副本所做的任何更改都不会反映。我不知道如何克服这一点。

Mark Simms 早在 2010 年就写了一篇关于在 StreamInsight 中使用参考流的博客文章，解释了如何使用静态数据源，并表示他的下一篇文章将描述使用 SQL Server。

不幸的是，那个帖子从未发生过。

编辑：我已经更改了这篇文章中的课程以匹配 Mark Simms 帖子中的课程，并尝试整理并详细说明我的过程。

我有一个IQStreamable使用.DefineObservableMicrosoft.ComplexEventProcessing.Application

代码看起来很正常，但我不明白的是，当我使用Task.Run()into传入的参数时DefineObservable，出现异常。

但是，当我直接使用该属性而不将其传递给内部方法Task.Run()时，它起作用了。

例外

An unhandled exception of type 'System.InvalidOperationException' occurred in Microsoft.ComplexEventProcessing.Diagnostics.dll

Additional information: Cannot serialize value of type 'System.IObservable'1[ValueObjects.Price]'.

方法

来电：

问题：

1. StreamInsight 是否序列化观察者对象？为什么？

2. 之间有什么不同

这 .DefineObservable<Price>(() => PriceObservable) 和

这个 DefineObservable<Price>(() => PriceRealtimeProvider.Instance.PriceObservable)

为什么使用参数会导致问题？

我不知道为什么我的 StreamInsight 引擎不能处理超过 140 个事件/秒，在每秒传递超过 8000 个事件的条件下。我在性能监视器中看到我的查询的事件数。并且在应用程序的控制台中就像引擎避免了很多事件。我有一个 id 为 200 的事件，然后是下一个 id 为 3330 的事件。有人知道问题出在哪里吗？

对于我的测试，我有一系列查询。第一个查询的输出是第二个查询的输入，依此类推，对于这种情况，结果不超过 140 个事件/秒。

现在我用一个简单的查询测试了我的应用程序，它输出从输入流接收到的所有事件，在这种情况下，服务器处理大约 2000 个事件/秒。我有一些带有结果的图像，但不幸的是我还不能放它们。困扰我的是为什么事件/秒的数量突然减少到 0 以及为什么引擎仍然没有考虑所有输入事件。

这是我的服务器配置和我正在使用的查询。

接下来，当我尝试运行我的最后一个查询（从链中）时，我得到了大约 1500 个事件/秒。仍然存在事件/秒突然减少的问题。我认为对事件进行的每次转换都会产生一个需要由引擎处理的新转换。因此，来自查询链的事件数应该超过 1500。如果我错了，请纠正我。我是这个领域的新手，欢迎任何建议。

我认为问题出在下面的课程中。我也尝试使用PointEvent<SendorDataEvent>而不是SensorDataEvent，并尝试插入 CTI，但没有结果。

谢谢你。

I'm working on porting over some Reactive Extension queries to StreamInsight but have run into an issue with an overlapping window query.

I have a source setup in my StreamInsight server and I'm trying to write an overlapping window query like this:

where source is IQbservable<EventPattern<MyEventArg>> and query is then IQbservable<IObserverable<EventPattern<MyEventArg>>>

With Reactive the observer was created as follows:

How can I attach an observer to retrieve the equivalent output from StreamInsight?

我已经工作StreamInsight了几个星期了，通常让它做我想做的事，但有一个例外。我不能让多个查询在同一个进程中运行。例如，假设eventStream是IQStreamable<eventClass>：

当它运行时，只有大约一半的 query1 事件被发送到query1Observer，并且只有大约一半的 query2 事件被发送到query2Observer。其他人就这样消失了。如果我只是做我的绑定

然后将所有查询的结果通知给观察者——但当然只针对 query1。如何运行 2 个查询，通知同一进程中的不同观察者？

我是否遗漏了什么，或者流洞察是一种将传统 BI 解决方案与 SSAS/SSRS/SSIS 结合起来的更简单方法。

传统的：

流洞察：

在传统方式中，我们没有实时分析的能力，您总是需要定义一个时间段来再次运行 SSIS 包。但是，我们可以使用 CDC（更改数据捕获），这使其具有一定的实时性。CDC 检查数据库中的更改并触发 SSIS 包。

StreamInsight 提供实时 ETL 解决方案。

这是正确的吗？

在 StreamInsight 2.1 中，我尝试使用用户定义的函数来扩展功能。目前，所有工作都在 IDE 中。

流处理中的段

这很好，一切运行正常，但它不会让我使用 f11 或断点进入 UDF。有没有一种简单的方法来查看 UDF 的阴谋？