Esper技术培训教程

Esper技术培训内容事件和事件处理Esper简介及架构输入、输出适配器EsperAPI二次开发一、事件和事件处理什么是事件？事件的基本特征事件举例简单事件处理(SEP):SimpleEventProcessing事件流处理(ESP):EventStreamProcessing复合事件处理(CEP):ComplexEventProcessing事件(Event)是有意义的状态变化asignificantchangeinstate例如：股票价格的变化密码变更最后一次服务的响应时间10kV配网故障的一次告警事件在系统中的表述XML、POJO、Map(Key-value)等1.1事件是什么？不只是“发生什么事情”发生事情的不可变记录事件要素：标识、发生时间、有意义的属性事件间可能存在某种关联：时间顺序、因果关系、聚合关系、依赖关系等1.2事件的基本特征移动社交—(如：旁街网)位置信息用户标识用户位置所在社区信息变更用例根据用户位置变更，定向推送所在区域的服务信息：商城、饭店、电影院、银行、公交站等信息统计和分析用户的日常行为规律1.3事件举例基于单个事件单个事件触发并响应通常采用JMSQueue:点对点Topic：发布和订阅1.4简单事件处理

基于”流”的处理单个事件不会触发“反映”需要分析事件流各种滑动窗口基于时间基于事件数量1.5事件流处理什么是复合事件处理？

先看一些生活化的例子。其实你每天都在做人肉CEP，只是你不知道而已。所谓人肉CEP，就是通过各种感知器官，将感觉汇集到大脑，经过综合判断得到一个更具有意义的事件。在这个事件被判定出来之后，你可能会想要为此采取必要行动。

个人版：

皮肤感觉温度下降是一个基础事件（Base

Event）或称简单事件（Simple

Event）。耳朵听到远方传来的雷声、眼睛看到闪电，都是基础事件。集合以上基础事件，得到一个复合事件：快要下雨了。于是，你考虑到需要：收衣服或者出门要带伞。

企业版：

身为公司领导，你看到员工经常有一堆事没做完，居然还在上班时间上网偷菜。于是你得到一个复合事件，就是：这是一个不负责任的员工。你的处理方式就是将他辞退。

家庭版:

外面的饭店又再涨价，而且常常出去吃饭，有时候一不小心经常吃出”怪物”，每天听新闻又在说地沟油、苏丹红…等饮食安全问题，通过这些事件后。你考虑需要，每天上班带饭比较安全。

看了这些例子，现在应该能了解CEP系统的大致作用了：先捕获各种细微的事件（基础事件），然后通过分析整理，找出更有意义的事件（复合事件），最后决定采取什么行动。其中事件的分析整理以找出更有意义的事件，正是CEP的核心，也是最困难的地方。1.6复合事件处理CEP

看了这些例子，现在应该能了解CEP系统的大致作用了：先捕获各种细微的事件（基础事件），然后通过分析整理，找出更有意义的事件（复合事件），最后决定采取什么行动。其中事件的分析整理以找出更有意义的事件，正是CEP的核心，也是最困难的地方。复合事件：大量由其他事件触发的事件基于事件流的处理需要对多个事件(流)做复杂的分析分析、发生模式过滤聚合相关事件关联事件关联因果关联空间关联1.6复合事件处理3.1事件在CEP中的位置事件源事件消费方事件处理总线事件缓存规则驱动事件流处理特殊算法\操作分析状态引擎规则引擎流程查询神经网络等状态模型基础规则流查询参数大量的实时数据作为事件源接入事件处理总线，CEP引擎通过指定的规则，处理这些实时数据和缓存的历史数据，并通过事件处理总线将有意义的事件提供给事件消费方。1.1在电力CEP应用场景中理解事件事件源1事件源2

事件源3

事件源4

事件源5处理有意义的事件BPM复杂事件处理总线(CEPBUS)：将电网企业的所有实时信息数据作为事件接入到复杂事件处理总线上复杂事件处理器(CEPAgent/Processor)：是复杂事件处理的一个功能单元，可以拆分事件和合并事件，也可以根据规则发现事件，在处理事件时，可以有任意多个处理器组合处理事件事件发现后，交给业务处理单元，复杂事件处理(ECP)，完成事件发现、事件处理、结束处理一套流程对于有意义的事件，需要启动相应的业务流程处理发现有意义事件：复杂事件处理引擎根据电网业务规则，在不同事件源中找出相关的组合事件。这种出现在不同的事件源中，而相互又有一定关系的组合事件，定义为有意义事件复杂事件处理引擎(CEPEngine)：复杂事件处理的核心，运行复杂事件处理器。主要任务是组合或分解事件，根据规则最终发现有意义的事件二、Esper技术简介IT领域的CEP现状商业CEP产品Esper简介Esper架构业界CEP应用

这是一个IT事件爆发的时代，各种IT系统之间或系统内部，每天产生大量事件。系统在关键点输出日志、系统之间交流信息，都是事件。但我们对这些事件往往视而不见，当成垃圾一股脑儿全扔了。其实只要处理得当，垃圾也可以变成资源。2.1IT领域的CEP2.2商业CEP产品2.3Esper简介

基于Java的CEP/ESP容器轻量级、可嵌入开源包括CEP和ESP项目背景商业支持被广泛集成到商业产品:WeblogicEventServer具体的说Esper是一个Java开发并且开源的轻量级和可扩展的事件流处理和复合事件处理引擎。复合事件处理(CEP:ComplexEventProcessing)：是一种实时事件处理并从大量事件数据流中挖掘复杂模式的技术。事件流处理ESP(ESP:EventStreamProcessing)：是一种从大量事件数据流中过滤，分析有意义的事件，并能够实时取得这些有意义的信息的技术。2.4Esper架构引擎(Engine):独立单元(时间、线程、事件流)基本语句(Statements):EventProcessingLanguage(EPL)事件处理器(Listener):简单的Java接口2.5业界CEP应用三、输入、输出适配器输入、输出适配器在Esper中的位置输入适配器输出适配器3.1输入输出适配器所处的位置输入输出适配器，API提供对实时数据流输入到Esper容器，或事件输出的各种途径或方式。在这个过程需将实时数据流转成实时事件流，并以Object、Map、Node形式的事件发送到引擎。CVSSpringJMSHTTPSocket关系型数据库3.2输入适配器SpringJMSHTTPXMLJSON3.3输出适配器四、EsperAPI及Esper引擎EsperCEP和事件流分析CEP和关系型数据库专注于CEP的Esper引擎EsperAPI概述EsperSimple事件的表达事件的发送ESP/CEPStatements(EPL)处理模式EQL(EPL)ESP查询CEP查询EQL/EPL高级查询Esper引擎是为了满足对事件进行分析并做出反应等这些应用需求而产生的。这些应用要求实时或接近实时处理事件（或消息）。有时候是为了应对复杂事件处理(CEP)和事件流分析的。关键要考虑这些类型应用的（高）吞吐量、（低）响应时间和需求逻辑的复杂程度（复杂计算）。高吞吐量-应用处理大量的事件(或消息)，每秒1000至100K之间的消息；低响应-应用实时响应并处理(几毫秒到几秒)复杂计算-应用检测事件(事件关联)之间的模式，过滤事件，触发事件等。Esper引擎的设计，使其更容易建立和扩展CEP应用。4.1EsperCEP和事件流分析关系型数据库其中的标准查询语音(SQL)是专为大多数数据相对静态和复杂查询不太频繁的应用，此外，大多数数据库将所有数据(内存数据库除外)都存储在磁盘上，因此，需优化磁盘的访问。为了从数据库检索数据，应用程序必须发出一个查询操作。如果应用程序需要查询10次每秒的数据，那么它必须发出每秒10次的查询操作，这样不是很好的扩展到数百或数千每秒的查询。数据库触发器可以被用来在相应数据库更新事件时触发，然而，数据库触发器往往是比较缓慢，不能够轻松的执行复杂的条件检查和处理逻辑并响应。内存数据库比传统关系型数据库可能更适合CEP应用，因为他们一般具有良好的查询性能。然而，他们没有提供实时的查询结果优化，需要CEP和事件流分析。4.2CEP和关系数据库Esper引擎的工作原理有点想倒置的数据库，而不是存储数据和运行对存储数据的查询，Esper引擎允许应用程序存储查询并运行数据通过，来代替存储数据并执行查询存储数据的工作方式。从Esper引擎的响应是实时的，当条件发生就立即查询，因此执行模式是连续的，而不是只有当一个查询提交时才执行。Esper提供了两种机制来处理事件：事件模式和事件流查询。Esper提供了一个事件模式语言来指定基于表达式的事件模式匹配，这个模式匹配引擎是通过一个状态机来实现的。这个事件处理的方法匹配期望存在的队列或者不存在的事件或者事件的组合。它包括以时间为基础的各个事件之间的关系。Esper还提供事件流查询，这样可以是事件流分析CEP应用的需求。事件流查询提供窗口、聚合、连接和分析的函数来处理事件流。这些查询是通过EPL语句来实现的。EPL已经被设计成和SQL查询语言相似，但使用意义与SQL不同。EPL用于将需要构造的数据放入到一个事件流中并驱动数据的流动。在数据流动的过程中对数据进行处理，来得到我们最后所需的结果。4.3专注于CEP的Esper引擎EPServiceProvider引擎线程事件流EPStatementStatement/Queries事件查询语音：EQLUpdateListener:ListenerPOJI4.4API概述事件类OrderEvent.java4.5EsperSimpleListenerUpdateListener接口实现类OrderListener.java4.5EsperSimple测试类APP,进行事件的发送4.5EsperSimple输出结果：10.015.020.025.030.035.040.045.050.055.0POJOKey-value对(java.util.Map)XML(org.w3c.dom.Node)通过各种输入适配器将数据流最终封装成这三种对象的事件流，并发送到引擎进行处理。发送事件(sendEvent)也只提供这三种对象的传入如上面Esper示例中传入了OrderEvent事件对象4.6事件的表达如EsperSimple中4.7事件的发送如EsperSimple中4.8ESP/CEPStatements(EPL)连续处理结果集发生改变时通知Listener新的事件达到旧事件超出结果集/范围内建数据库-进行事件缓存4.9处理模式类似SQL语法流Streams：表事件Event：记录事件属性EventAttributes:记录字段查询Query:ESP查询CEP查询4.10EQL(EPL)单个事件查询：select*fromOrderEvent滑动式窗口:时间窗口范围内的查询：

selectAvg(price)fromOrderEvent.win:time(5sec)指定数量窗口范围内的查询： selectAvg(price)fromOrderEvent.win:length(5)跳跃式窗口：即批量处理：在通知Listener前累积事件，然后一次通知基于时间的累积：selectAvg(price)fromOrderEvent.win:time_batch(5)基于长度的累积：selectAvg(price)fromOrderEvent.win:length_batch(5)4.11ESP查询(1)滑动式扫描窗口

可以以时间(time)、长度(length)、分类(sorted)、分等级(ranked)、累积(accumulating)、时间顺序(time-ordering)，外部定时(externally-timed)，基于表达式的事件来设定扫描方式和区间。4.11ESP查询(2)跳跃式扫描窗口

可以以事件、长度或多策略组合基于表达式的事件来设定扫描方式和区间。4.11ESP查询(3)4.11ESP查询(3)模式关键字定义事件匹配模式标识复合事件流CEP查询示例4.12CEP查询周期性/重复性:every逻辑操作：or、and、not跟着发生：_>子查询条件表达式：timer:within5秒内所有A或B事件:(AorB)wheretimer:within(5sec)定时来监视事件的发生timer:intervalA事件后等待10秒：A_>timer:interval(10seconds)timer:at每五分钟：everytimer:at(5,*,*,*,*)4.12.1CEP查询-模式关键字事件集合Events:A1

B1C1B2A2D1A3B3E1A4F1B44.12.2CEP查询-事件匹配模式与标识复合事件流(1)模式1 [everyA_>B]子表达式：{A1,B1},{A2,B3},{A3,B3},{A4,B4}模式2 [every(A_>B)]子表达式：{A1,B1},{A2,B3},{A4,B4}事件匹配模式与标识复合事件流(1)事件集合Events:A1

B1C1B2A2D1A3B3E1A4F1B44.12.2CEP查询-事件匹配模式与标识复合事件流(2)模式3 [A_>everyB]子表达式：{A1,B1},{A1,B2},{A1,B3},{A1,B4}模式4 [everyA_>everyB)]子表达式1：{A1,B1},{A1,B2}子表达式2：{A1,B3},{A2,B3},{A3,B3}子表达式3：{A1,B4},{A2,B4},{A3,B4}子表达式4：{A4,B4}电力设备温度传感器检测采样样本Simple：传感器标签名称、温度、设备标识需求：探测温度，温度过高就警告警告条件：同一个温度传感器在90秒内，连续检测到温度超过50度。4.12.3CEP查询示例(1)everysample=Simple(temperature>50)_>( ( Simple(tagName=simple.tagName,temperature>50)andnot Simple(tagName=simple.tagName,temperature<=50) ) _> ( Simple(tagName=simple.tagName,temperature>50)andnot Simple(tagName=simple.tagName,temperature<=50) ))4.12.3CEP查询示例(2)事件过滤过滤掉温度在0-50范围的事件Select*fromsimple(temperaturein[0,50])滑动窗口和聚合统计30秒内温度为50的事件Selectcount(*)fromsimple(temperature=50).win:time(30sec)分组窗口和聚合每一分钟统计一次10分钟内，各个温度所出现的频率Selecttemperature,count(*)fromsimple.std:groupwith(‘temperature’).win:time(10min)outputevery1min4.13EsperEQL\EPL高级查询JoinsandouterJoins假如