标签: complex-event-processing

我在Steam上玩Team Fortress 2时注意到的成就的复杂性是惊人的.

有没有资源描述他们如何构建他们的成就引擎？有许多数据必须同时并实时处理,以检测成就标准的匹配.我觉得它不会使用熟悉的概念,例如"复杂事件处理".

寻找一种使用Esper(EPL)语句在5 EMA5和EMA20窗口上计算指数移动平均线的方法.

我有一个priceEvent(timeStamp,symbol和price)流,我在5的滑动窗口上写了一个简单的移动avrage SMA.但对Esper来说还是一个新手,正在寻找一种在滑动窗口上计算指数移动平均线(EMA)的方法.

http://www.iexplain.org/ema-how-to-calculate/

如果有人可以帮助我编写抛物线SAR功能,那将会很有帮助

场景:最终用户希望定义对系统内发生的一组数据更改事件的操作.这些操作应该是可配置的.我对此有几点疑问:

1. 我的第一个问题是我们如何监控数据更改并从数据库层提升事件？因为没有单点输入数据库.我所知道的可能的解决方案是DB触发器,更改跟踪和更改数据捕获.它们真的是很好的选择,因为它们都是基于Pull模型的吗？

2. 谁应该观察这些事件？我们应该将它们排成一张桌子并从那里逐一挑选吗？

3. 它是复杂事件处理的正确用例吗？

4. 很明显,动作的可配置性可以直接来自任何规则引擎,如drools,flexrule等.这是正确的方法吗？有很多博客都在谈论规则引擎不好选择.

我正在考虑使用Drools Fusion来实现以下示例性规则:

• 如果顾客在一年内购买了两件相同的产品,她将免费获得第三件
• 为了在最多连续三次交易中花费超过200美元,客户可以获得折扣
• 当客户闲置超过一年时,会生成提醒
• 当单个订单大于最后六个monhts内的平均订单价值时,将通知主管
• ......等等,这些只是我头脑中的例子

所有这些规则都可以使用Drools Fusion轻松表达.然而,在创建原型并查看文档后,似乎该产品更适合短期,瞬态事件(股票市场,实时决策).

问题是:是否需要将所有事件存储在内存中？(可能有数百万的旧购买).此外,Drools Fusion可以存储事件,以便它们在服务器重启后能够存活吗 (理想情况下在数据库中).

我知道Drools Fusion事件只是Drools Expert中的事实,所以也许这个问题可以扩展到:Drools Expert能否坚持并懒惰加载事实？

我正在开发一个接受100k +唯一输入的应用程序 - 为简单起见假设每个输入都是浮点值(a,b,c ......等) - 尽管它们也可以是字符串等.应用程序有很多规则/与这些输入相关的事件/触发器.

例1:

Rule[(a > b) and (c <= d)] --> [execute EventX]

定义1:上述规则说:当输入'a'的值大于'b'的值并且输入'c'的值小于或等于'd'的值时,执行EventX

例2:

Rule[x != x.prev] --> [execute EventZ]

定义2:上述规则说:如果值更新后,如果输入'x'的当前值不等于其先前值(值已更改).执行EventZ

我发现,随着输入数量的增加,以及规则数量的增加,评估"特定"规则并确定是否应该触发事件所需的总计算量正在失控 - 投掷该问题的速度越来越快,而且没有按预期进行扩展.

目前,在每次新的输入更新时,相关的输入/变量在哈希表中查找,该哈希表将变量映射到包含它的规则.随后评估每个规则,如果结果为真或可操作,则异步触发相关事件.

这个问题出现在复杂事件处理领域,遗憾的是,这个领域中的大多数架构使用了上面描述的相同的死角方法 - 可能还有一些与评估/重新评估事物的频率有关的改进.我们的想法是拥有一个能够近乎实时地做出反应的系统.在多个节点上分配规则和输入似乎也不能很好地工作,因为在某些情况下,超过95%的活动规则中存在少数输入.

我希望是否有任何其他SO'ers,知道更好的方法来解决这个问题,数据/结构或算法.

我想到的一个简单的想法是,可以构建一个依赖逻辑推理的列表.

如果有两个规则是这样的:

Rule[a < b] --> [exec E1]
Rule[b >= a] --> [exec E2]

然后对'a'或'b'的更新不应该要求对两者等进行评估,但我发现构建这样的逻辑推理结构非常复杂且容易出错,并且难以完全和严格地测试.

输入可以代表股票价格,温度传感器等.

还要注意,一些输入在时间上受到限制,这意味着规则可能要求变量的状态在特定位置/状态下持续一段时间(例如:MSFT的价格>最后30秒的20美元),目前,这是通过使用值为0或1/false或true的'代表变量'(facade)来实现的(变量的值由单独的机制决定,这通常是规则被触发的结果).

还应该注意的是,由于近实时约束和每秒产生的数据量,使用具有触发器和存储过程的DB的选项是不可能的.

Actor 模型框架（例如Orleans）和复杂事件处理（CEP）（例如Apache Storm）之间有什么区别？

这些方法中的每一种都适用（或者一种方法比另一种方法更适用）的用法示例肯定会有所帮助。

我们正处于构建物联网云平台项目的初期。有一些众所周知的部分可以实现完整的物联网平台解决方案。其中之一是实时规则处理/引擎系统，它需要了解流事件与最终用户以可读格式（SQL 或 Drools if/when/then 等）动态定义的任何规则相匹配。

我很困惑，因为互联网上有很多产品、项目（Storm、Spark、Flink、Drools、Espertech 等），所以考虑到我们有 3 人的开发团队（初级、中高级、高级），什么是最好的选择？

• 选择 Apache Flink 等流媒体项目之一并学习好？
• 选择一个完整的解决方案（AWS、Azure 等）

我需要知道如何通过事件发布者从Enteprise Integrator到流处理器发布统计信息。

我在EI上执行了事件发布者的以下实现

<?xml version="1.0" encoding="UTF-8"?>
<eventPublisher name="MessageFlowStatisticsPublisher"
  statistics="enable" trace="enable" xmlns="http://wso2.org/carbon/eventpublisher">
  <from streamName="org.wso2.esb.analytics.stream.FlowEntry" version="1.0.0"/>
  <mapping customMapping="disable" type="wso2event"/>
  <to eventAdapterType="wso2event">
    <property name="username">admin</property>
    <property name="protocol">thrift</property>
    <property name="publishingMode">non-blocking</property>
    <property name="publishTimeout">0</property>
    <property name="receiverURL">tcp://xxx:7611</property>
    <property encrypted="true" name="password">xxx</property>
  </to>
</eventPublisher>

在流处理器上，我有一个简单的siddhi应用程序，用于接收数据并将其打印到日志中，如下所示

@App:name("FlowEntryApp")
@App:description("Plan of flow entry")

@source(type='wso2event', @map(type = 'wso2event'))
define stream FlowEntry(compressed bool, tenantId int, messageId string, flowData string);

@sink(type='log', prefix='My flowEntry:')
define stream TestOutputFlowEntry(messageId string, flowData string);

@info(name='FlowEntryOutput')
from FlowEntry
select messageId, flowData
group by messageId
insert into TestOutputFlowEntry;

另外，我将发布统计信息的所有配置都设置为代理服务的“启用统计信息”和“启用跟踪”。当我调用服务时，eventPublisher将wso2event发送到SP，这可以正常工作。但是在SP端，SP处理错误“缓存中存在streamId org.wso2.esb.analytics.stream.FlowEntry：1.0.0的没有StreamDefinition”

我知道，这个问题出在siddhi应用程序中，我定义了流“ FlowEntry”而不是“ …

Rx中缺少哪些功能在CEP平台中,例如NEsper或MS StreamInsight？

当在一定时间内未从流中接收到任何数据时，我该如何使用Flink的DataStream API实现一个操作符来发送事件？