Esper性能问题

Question

我们有一个esper运行的原型，但性能相当缺乏。我想这在某种程度上是我的错，而不是esper的固有问题，所以在寻找我的性能问题所在时，我一直在寻求帮助。

我正在运行esper服务的一个实例，并按照以下方式分配内存约束：-Xmx6G -Xms1G (我尝试了这些值的各种组合)。并且它可以使用4个核心的CPU。在这些测试时，没有其他服务在运行，只有esper，kafka，动物园管理员。

我使用Akka流将事件流到Esper中，服务非常简单，它从kafka流进来，将事件插入Esper，Esper有3 EPStatements测试和工作。有一个监听器，我将它添加到所有三个语句中，监听器将匹配的事件输出到kafka。

有些事情我试图将性能问题隔离开来：

1. 删除一些EPStatements
2. 删除所有EPStatements
3. 移除侦听器
4. 删除EPStatements和侦听器
5. 删除esper .sendEvent(.)(这大大提高了性能，因此似乎是esper问题，而不是akka问题)

只有上面的第4位产生了任何显著的可观察的性能效益。

下面是我们正在通过esper运行的一个示例查询。它是经过测试和工作的，我已经阅读了文档的性能调优部分，在我看来，这似乎还可以。我的所有查询都采用类似的格式：

select * from EsperEvent#time(5 minutes)
  match_recognize (
    partition by asset_id
    measures A as event1, B as event2, C as event3
    pattern (A Z* B Z* C)
    interval 10 seconds or terminated
    define
      A as A.eventtype = 13 AND A.win_EventID = "4624" AND A.win_LogonType = "3",
      B as B.eventtype = 13 AND B.win_EventID = "4672",
      C as C.eventtype = 13 AND (C.win_EventID = "4697" OR C.win_EventID = "7045")
)

一些密码..。

这是我的阿克卡溪流：

  kafkaConsumer
    .via(parsing) // Parse the json event to a POJO for esper. Have tried without this step also, no performance impact
    .via(esperFlow) // mapAsync call to sendEvent(...)
    //Here I am using kafka to measure the flow throughput rate. This is where I establish my throughput rate, based on the rate messages are written to "esper_flow_through" topic.
    .map(rec => new ProducerRecord[Array[Byte], String]("esper_flow_through", Serialization.write(rec)))
    .runWith(sink)

esperFlow (默认情况下并行度=4)：

val esperFlow = Flow[EsperEvent]
    .mapAsync(Parallelism)(event => Future {
      engine.getEPRuntime.sendEvent(event)
      event
    })

听众：

  override def update(newEvents: Array[EventBean], oldEvents: Array[EventBean], statement: EPStatement, epServiceProvider: EPServiceProvider): Unit = Future {
    logger.info(s"Received Listener updates: Query Name: ${statement.getName} ---- ${newEvents.map(_.getUnderlying)}, $oldEvents")
    statement.getName match {
      case "SERVICE_INSTALL" => serviceInstall.increment(newEvents.length)
      case "ADMIN_GROUP" => adminGroup.increment(newEvents.length)
      case "SMB_SHARE" => smbShare.increment(newEvents.length)
    }
    newEvents.map(_.getUnderlying.toString).toList
      .foreach(queryMatch => {
        val record: ProducerRecord[Array[Byte], String] = new ProducerRecord[Array[Byte], String]("esper_output", queryMatch)
        producer.send(record)
      })
  }

业绩观察：

• 输入流的速率为~2.4k /秒。
• 我们看到esper从一开始就无法跟上。以每秒600度最大
• Esper的吞吐量逐渐下降
• 最终，esper吞吐量稳定在<100 /秒

​

​

​

侧写，这里似乎没有什么不对劲的地方：

​

​

这个速率看起来很低，所以我假设我在这里遗漏了一些关于esper配置的东西？

我们的目标吞吐量是每秒10K左右。我们离这个目标还有很长的路要走，而且我们在星火区也有一个类似的POC，离这个目标越来越近。

更新：

在@user650839注释之后，我能够将吞吐量提高到每秒1k。这两个查询产生相同的吞吐量：

select * from EsperEvent(eventtype = 13 and win_EventID in ("4624", "4672", "4697", "7045"))#time(5 minutes)
     match_recognize (
       partition by asset_id
       measures A as event1, B as event2, C as event3
       pattern (A B C)
       interval 10 seconds or terminated
       define
         A as A.eventtype = 13 AND A.win_EventID = "4624" AND A.win_LogonType = "3",
         B as B.eventtype = 13 AND B.win_EventID = "4672",
         C as C.eventtype = 13 AND (C.win_EventID = "4697" OR C.win_EventID = "7045"))

create context NetworkLogonThenInstallationOfANewService
start EsperEvent(eventtype = 13 AND win_EventID = "4624" AND win_LogonType = "3")
end pattern [
 b=EsperEvent(eventtype = 13 AND win_EventID = "4672") ->
 c=EsperEvent(eventtype = 13 AND (win_EventID = "4697" OR win_EventID = "7045"))
 where timer:within(5 minutes)
]

context NetworkLogonThenInstallationOfANewService select * from EsperEvent output when terminated

然而，每秒1k的速度仍然太慢，无法满足我们的需要。

Answer 1

匹配识别是模糊的.A事件或B事件或C事件事件也可以是Z事件，因为任何东西都与Z事件匹配(Z是未定义的)。因此，有大量的组合是可能的。我认为，对于4个即将到来的事件，已经有像1*2*3*4组合，比赛识别保持跟踪！匹配识别跟踪所有可能的组合，当物料匹配时，识别排序和排序，并输出所有/任意/部分。在这里，匹配识别可能是一个糟糕的选择，或者可以将Z定义为与A/B/C不匹配的东西。

与匹配识别不同，我将使用一个上下文，该上下文启动一个A事件，并以一个C事件终止，“终止时输出”。

另外，他们设计查询的方式--时间窗口将保留所有事件。你可以做得更好。

select * from EsperEvent(eventtype = 13 and win_EventID in ("4624", "4672", "4692", "7045"))#time(5 minutes) match_recognize ( ......... define A as A.win_EventID = "4624" AND A.win_LogonType = "3", B as B.win_EventID = "4672", C as C.win_EventID = "4697" OR C.win_EventID = "7045" )

注意，EsperEvent(eventtype=13 ....)在事件进入时间窗口之前丢弃它们。文档中有一个性能提示，它使用筛选条件来删除不需要的事件。

编辑:一个错误是测量IO吞吐量和Esper吞吐量作为一个整体。移除IO。使用您的代码生成的数据使用Esper测试Esper。一旦自信，将IO添加回。