Esper epl语句详解

EPL，Event Process Language，事件处理语言。类似于SQL，描述了esper要分析的内容。

统计窗口

以下为常用窗口。

win:length(size)//攒够size条数据后触发UpdateListener()函数。滑动窗口，攒满之后新来一个移除一个，并触发。
win:length_batch(size) //攒够size条数据后触发，并清空队列。再攒满了再触发。
win:time(time period)//第一次触发在period秒后，然后每一秒触发一次。
win:time_batch(time period) //每period秒触发一次。
win:keepall()//无参数，记录所有进入的数据，除非使用delete操作，才能从窗口移出数据。
std:unique(criteria)//对不同的criteria[例如:id]保留其最近的一条事件

字符串操作

与标准SQL类似。支持 like '%' 这样的模糊查询。

不支持 LENGTH(str)，可以用 string.length() 函数替代，见下行。

select * from appTable.win:time(5 sec) as a where a.price.length() < 3

注解

@注解名字(注解内容)
常用的有 @Hint，用于限制Group by的生存时间，使虚拟机能及时回收内存。这两个属性分别为reclaim_group_aged和reclaim_group_freq。例子见下：

// 根据color对10秒内进入的Apple事件进行分组计算平均price。对8秒内没有数据更新的分组进行回收,每2秒回收一次
@Hint('reclaim_group_aged=8，reclaim_group_freq=2')select avg(price) as aPrice, color from Apple.win:time(10 sec) group by color

@Priority，指定EPL的优先级，参数只有一个，为整数，可负可正。例如：@Priority(10)。

连接

// 当老师的id和学生的id相同时，查询学生的姓名和老师的姓名
select s.name, t.name from Student.win:time(10) as s inner join  Teacher.win:time(10) as t on s.id=t.id

类似的，esper也支持left outer join。

Unidirectional Join

join的事件都需要data window或者view修饰，目的是为了暂存事件以便等待满足条件的事件并执行join。如果想让某个事件到来时直接触发join，不需要暂存，也就是不需要data window或者view修饰，则需要加上一个特殊关键字——unidirectional。

output

Output是EPL中非常有用的东西，用来控制Esper对事件流计算结果的输出时间和形式，可以以固定频率，也可以是某个时间点输出。

// 30分钟内，每进入一个OrderEvent，统计一次sum price，并且每60秒输出一次统计结果。
select sum(price) from OrderEvent.win:time(30 min) output snapshot every 60 seconds
// 30分钟内，每进入一个OrderEvent，统计一次sum price，并且每60秒输出第一次的统计结果。
select sum(price) from OrderEvent.win:time(30 min) output first every 60 seconds

insert into

把一个事件流的计算结果放入另一个事件流，然后可以对这个事件流进行别的计算。所以Insert into的一个好处就是可以将是事件流的计算结果不断级联，对于那种需要将上一个业务的结果数据放到下一个业务处理的场景再适合不过了。

//insert into 示例
String epl3="create schema UBTInfo as (`ip` string, `url` string, `userAgent` string)";
String epl4="create schema UBTInfoCounter as (`ip` string, `counts` long)";
String epl5="insert into `UBTInfoCounter` select table1.ip as ip,count(*) as counts from `UBTInfo`.win:time_batch(10 sec) as table1 group by table1.ip";

注意：以上面的代码片为例，若不执行epl4，执行epl5也不会报错，因为要插入的schema不必先定义！

insert into也可以绑定listener。可以在实现接口的方法中调用 newEvent.get("_attribute")。
insert into A from B后，B中盛放事件的队列不受影响。