esper学习之七：epl语法(三)

国庆七天，本想出去玩玩，可是哪里都是人，所以还是家里蹲吧。上篇说到了 Select Clause和 From Clause，今天这篇就说说 Aggregation，Group by，Having 和 Output Clause。先预告一下，由于例子比较多，所以篇幅会有 些长，需要各位耐心观看哦。 1.Aggregation 和 SQL一样，EPL 也有 Aggregation，即聚合函数。语法如下： plain view plaincopy 1. aggregate_function(all|distinct expression) aggregate_function就是聚合函数的名字，比如 avg，sum 等。expression 通 常是事件流的某个属性，也可以是不同事件流的多个属性，或者是属性和常量、 函数之间的运算。举例如下。 plain view plaincopy 1. / 查询最新 5秒的 Apple的平均价格 2. select avg(price) as aPrice from Apple.win:time(5 sec) 3. 4. / 查询最新 10个 Apple的价格总和的两倍 5. select sum(price*2) as sPrice from Apple.win:length(10) 6. 7. / 查询最新 10个 Apple的价格，并用函数计算后再算平均值 8. select avg(Compute.getResult(price) from Apple.win:length(10) 函数只能是静态方法，普通方法不可用。即使是事件流里包含的静态方法，也 必须用“类名.方法名”的方式进行引用。 可以使用 distinct关键字对 expression加以约束，表示去掉 expression产生 的重复的值。默认情况下为 all关键字，即所有的 expression值都参与聚合运 算。例如： plain view plaincopy 1. / 查询最新 5秒的 Apple的平均价格 2. select avg(distinct price) as aPrice from Apple.win:time(5 sec) 3. 4. / 假如：5 秒内进入了三个 Apple事件，price 分别为 2,1,2。则针对 该 EPL的平均值为(2+1)/2=1.5。因为有 distinct的修饰，所以第二个 2不参与运算，事件总数即为 2，而不是 3。 以上就是聚合函数的使用方法，除此之外需要注意一下几点 1.聚合函数能用于 Select和 Having，但是不能用于 Where 2.sum,avg,media,stddev,avedev只能计算数值，至于 media，stddev 和 avedev代表什么意思，请自行百度。 3.Esper会忽略 expression为 null不让参与聚合运算，但是 count函数除外， 即使是 null也认为是一个事件。如果事件流集合 中没有包含任何事件，或者 包含的事件中用于聚合计算的 expression都是 null（比如收集 5秒内进入的 事件即为一个事件流集合），则所有聚合函数 都返回 null。 2.Group by Group by通常配合聚合函数使用。语法和 SQL基本一样，产生的效果就是以某 一个或者多个字段进行分组，然后使聚合函数作用于不同组的数据。简单语法 如下： plain view plaincopy 1. group by aggregate_free_expression , aggregate_free_expression , . 使用 Group by要注意一下几点： 1.Group by后面的内容不能包含聚合函数 2.Group by后面的内容不能是之前 select子句中聚合函数修饰的属性名 3.通常情况要保证分组数量有限制，以防止内存溢出。但是如果分组分了很多， 就需要使用Hint 加以控制。 2.1.Group by基本用法 针对上面的第三点，后面再说，先举几个例子说明下简单用法： plain view plaincopy 1. / 根据 color和 size来对 10个 Apple事件进行分组计算平均 price 2. select avg(price) as aPrice, color, size from Apple.win:length_ batch(10) group by color,size 该句子遵从 SQL的标准，如果某个事件的 color和 size和之前进入的事件的一 样，则归为一组，否则新建一组，并计算平均 price plain view plaincopy 1. / 根据 size来对 10个 Apple事件进行分组计算平均 price和 color 2. select avg(price) as aPrice, color, size from Apple.win:length_ batch(10) group by size 可以发现，group by 的对象只有 size，而 select中 color不聚合，则生成的 结果时，聚合函数会根据相同的 size分组进行平均 price的计算，但是 color 不是分组条件，所以 color有多少个就有多少组，即使存在一样的 color也不 会影响分组数量（实际上就是不分组），但一定记住，聚合函数还 是会根据分 组条件计算其修饰的属性。 plain view plaincopy 1. / 根 据 size来对 10个 Apple事件进行分组计算平均 price和 color 这一次 select子句中没有包含分组的字段 size，但是效果和上一个句子一样。 Esper仍然会根据相同的 size进行分组计算平均 price，只不过计算结果中只 有平均 price和 color，并且有十排结果。 plain view plaincopy 1. / 根 据 size乘 color来对 10个 Apple事件进行分组计算平均 price group by的对象只是一个值，以相同的值进行分组，所以上面和和普通的属性 字段一样，计算一个值进行分组。如果 group by后面的表达式值为 null，则 所有为 null的事件都被分为一组进行计算。但是如果使用了 count函数，则表 达式为 null的事件不会被计算在内。 2.2.Hint Hint是 Esper中注解的其中一个，如果不了解注解，可以先看看 Esper学习 之五：EPL 语法（一）的 第 7节再继续阅读Hint 的内容。之前对Hint 一笔 带过，那是因为它是专用于 Group by的。我们平时使用 Group by的时候，会 遇到分组数量太多的情况。比如以时间单位进行分组，那么内存使用一定是一 个大问题。因此Hint 为其设计了两个属性，用于限制 Group by的生存时间， 使虚拟机能及时回收内存。这两个属性分别为 reclaim_group_aged和 reclaim_group_freq reclaim_group_aged 该属性后面跟着的是正整数，以秒为单位，表示在 n秒内，若分组的数据没有 进行更新，则分组数据被 Esper回收。例如： plain view plaincopy 1. / 根据 color对 10秒内进入的 Apple事件进行分组计算平均 price,并 且对 5秒内没有数据更新的分组进行回收 2. Hint(reclaim_group_aged=5)select avg(price) as aPrice, color from Apple.win:time(10 sec) group by color /括号内可以使单引 号也可以是双引号 reclaim_group_freq 该属性后面跟着的是正整数，以秒为单位，表示每 n秒清理一次分组，可清理 的分组是 reclaim_group_aged决定的，也就是说要使用该参数，就要配合 reclaim_group_aged一起使用。可能不是很好理解，先看看例子： plain view plaincopy 1. / 根据 color对 10秒内进入的 Apple事件进行分组计算平均 price。 对 8秒内没有数据更新的分组进行回收,每 2秒回收一次 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 如果不使用 reclaim_group_freq属性，则默认值和 reclaim_group_aged的值一样，对上面来说就是回收的条件为 8秒内没有 数 据更新，且每 8秒回收一次。这样的话有可能出现这么一种情况，上一个 8秒 的某个分组在下一个 8秒还没到达时就已经持续 8秒没有数据更新了（这句话 会不会 有点绕？），但是必须等到回收的时间点到达时才能回收这个分组。在 分组产生很快的情况下，这样的回收不及时很可能会造成内存溢出。 reclaim_group_freq正是为这种情况做准备，回收的频率高一些，在一定程度 上能提高内存的使用率。 上面这两个属性的值除了可以使用正整数之外，也可以使用预先定义 的变量或者常量 3.Having Having的用法和 SQL一样，后面跟的是对聚合函数的计算结果进行过滤。 Where子句不能包含聚合函数，所以就由 Having来完成。示例如下： plain view plaincopy 1. / 根 据 size来对 10个 Apple事件进行分组计算平均 price和 color, 并且排除平均 price大于 5的分 组 s elect avg(price) as aPrice, color from Apple.win:length_batch(1 0) group by size having avg(price) 5 通常 Having配合 Group by使用，如果没有使用 Group by，那么就只有一组。 例如： plain view plaincopy 1. / 根 据 size来对 10个 Apple事件计算平均 price和 color,如果平均 price大于 5，则数据被排除 掉 s elect avg(price) as aPrice, color from Apple.win:length_batch(1 0) having avg(price) 5 Having后面可以跟多个判断式子，并且用 and，or 或者 not进行连接。例如： plain view plaincopy 1. / 根 据 size来对 10个 Apple事件计算平均 price和 color,如果平均 price大于 5并且平均 size小于 3，则数据被排除 掉 s elect avg(price) as aPrice, color from Apple.win:length_batch(1 0) having avg(price) 5 and avg(size) 4.Output 4.1.基本语法 Output是 EPL中非常有用的东西，用来控制 Esper对事件流计算结果的输出时 间和形式，可以以固定频率，也可以是某个时间点输出。简单语法如下： plain view plaincopy 1. output after suppression_def 2. all | first | last | snapshot every time_period | output_rat e events after suppression_def是可选参数，表示先满足一定的条件再输出。 all | first | last | snapshot表明输出结果的形式，默认值为 all。 every output_rate表示输出频率，即每达到规定的频率就进行输出。 time_period表示时间频率，相关语法在 Esper学习之五：EPL 语法（一）的第 2节有说到。output_rate events 表示事件数量。 举例说明如下： plain view plaincopy 1. / 30分钟内，每进入一个 OrderEvent，统计一次 sum price，并且每 60秒输出一次统计结果。 2. select sum(price) from OrderEvent.win:time(30 min) output snaps hot every 60 seconds 4.2.after 之前在讲解 Context的时候，有简单说到过 after。关于 Context，可参看 Esper学习之四：Context。after 在 output里的使用也很简单，语法如下： plain view plaincopy 1. output after time_period | number events . time_period表示时间段，number events 表示事件数量。表示从 EPL可用开始， 经过一段时间或者接收到一定数量的事件再进行输出。例如： plain view plaincopy 1. / 统计 20个 Apple事件的 sum price，并且在有 5个 Apple事件进入 后才开始输出统计结果 2. select sum(price) from Apple.win:length(20) output after 5 even ts 上面这个句子从第一个进入的事件进行统计，直到进入了 5个事件以后才输出 统计结果，之后每进入一个事件输出一次（这是 win:length的特性）。但是要 注意的是，after 之后的时间长度和事件数量会影响之后的时间或者事件数量。 什么意思？看个完整例子： java view plaincopy 1. /* 2. * 3. * author luonanqin 4. * 5. */ 6. class Banana 7. 8. private int id; 9. private int price; 10. 11. public int getId() 12. 13. return id; 14. 15. 16. public void setId(int id) 17. 18. this.id = id; 19. 20. 21. public int getPrice() 22. 23. return price; 24. 25. 26. public void setPrice(int price) 27. 28. this.price = price; 29. 30. 31. public String toString() 32. 33. return “id: “ + id + “, price: “ + price; 34. 35. 36. 37.class OutputAfterListener implements UpdateListener 38. 39. public void update(EventBean newEvents, EventBean oldEv ents) 40. 41. if (newEvents != null) 42. 43. int price = (Integer) newEvents0.get(“sPrice“); 44. System.out.println(“Bananas sum price is “ + price); 45. 46. 47. 48. 49.public class OutputAfterTest 50. 51. public static void main(String args) throws InterruptedEx ception 52. 53. EPServiceProvider epService = EPServiceProviderManager. getDefaultProvider(); 54. 55. EPAdministrator admin = epService.getEPAdministrator(); 56. 57. String banana = Banana.class.getName(); 58. / 统计最新 3个 Banana事件的 sum price，并且从 EPL可用 起，等待第一个事件进入后，以每两个事件进入的频率输出统计结果 59. String epl = “select sum(price) as sPrice from “ + bana na + “.win:length(3) output after 1 events snapshot every 2 eve nts“; 60. 61. EPStatement state = admin.createEPL(epl); 62. state.addListener(new OutputAfterListener(); 63. 64. EPRuntime runtime = epService.getEPRuntime(); 65. 66. Banana b1 = new Banana(); 67. b1.setId(1); 68. b1.setPrice(6); 69. System.out.println(“Send Banana Event: “ + b1); 70. runtime.sendEvent(b1); 71. 72. Banana b2 = new Banana(); 73. b2.setId(2); 74. b2.setPrice(3); 75. System.out.println(“Send Banana Event: “ + b2); 76. runtime.sendEvent(b2); 77. 78. Banana b3 = new Banana(); 79. b3.setId(3); 80. b3.setPrice(1); 81. System.out.println(“Send Banana Event: “ + b3); 82. runtime.sendEvent(b3); 83. 84. Banana b4 = new Banana(); 85. b4.setId(4); 86. b4.setPrice(2); 87. System.out.println(“Send Banana Event: “ + b4); 88. runtime.sendEvent(b4); 89. 90. Banana b5 = new Banana(); 91. b5.setId(5); 92. b5.setPrice(4); 93. System.out.println(“Send Banana Event: “ + b5); 94. runtime.sendEvent(b5); 95. 96. 执行结果： plain view plaincopy 1. Send Banana Event: id: 1, price: 6 2. Send Banana Event: id: 2, price: 3 3. Send Banana Event: id: 3, price: 1 4. Bananas sum price is 10 5. Send Banana Event: id: 4, price: 2 6. Send Banana Event: id: 5, price: 4 7. Bananas sum price is 7 由此可见，after 之后的 every子句要等到 after后面的表达式满足后才生效。 所以第一个事件进入后，every 2 events 生效，即等待两个事件进入后才输出 结果。对于时间也是要等到 after的子句满足后才开始计时。例如： plain view plaincopy 1. / 从 EPL可用开始计时，经过 1分钟后，每 5秒输出一次当前 100秒内 的所有 Banana的 avg price（即：第一次输出在 65秒时） 2. select avg(price) from Banana.win:time(100 sec) after 1 min sna pshot every 5 sec 4.3.first， last， all， snapshot 每当达到输出时间点时，可以用这四个参数来控制输出内容。下面分别介绍并 举例。 first 表示每一批可输出的内容中的第一个事件计算结果。比如： plain view plaincopy 1. select * from Fruit output first every 2 events 上面的句子表示每进入两个 Fruit事件，输出这两个事件的第一个。 last 和 first类似，表示每一批可输出的内容中的最后一个事件计算结果。比如： plain view plaincopy 1. select * from Fruit output last every 2 events 上面的句子表示每进入两个 Fruit事件，输出这两个事件的第二个，也就是最 后一个。 snapshot 表示输出 EPL所保持的所有事件计算结果，通常用来查看 view或者 window中 现存的事件计算结果。比如： plain view plaincopy 1. select * from Fruit.win:time(5 sec) output snapshot every 2 eve nts 上面的句子表示每进入两个事件输出 5 sec内的所有事件，且不会讲这些事件 从 5 sec范围内移除 all 也是默认值。和 snapshot类似，也是输出所有的事件，但是不同的是， snapshot相当于对计算结果拍了一张照片，把结果复制出来并输出，而 all是 把计算结果直接输出，不会复制。比如： plain view plaincopy 1. select * from Fruit.win:time(5 sec) output all every 2 events 上面的句子表示每进入两个事件输出 5 sec内包含的所有事件，输出的事件不 再保留于 5 sec范围内。 4.4.Crontab Output output的另一个语法可以建立定时输出，关键字是 at。语法如下： plain view plaincopy 1. output after suppression_def 2. all | first | last | snapshot at 3. (minutes, hours, days of month, months, days of week , seconds) minutes, hours, days of month, months, days of week , seconds这些都 是时间单位，语法后面再细说。举个简单的例子： plain view plaincopy 1. / 在 8点到 17点这段时间内，每 15分钟输出一次 2. select * from Fruit output at (*/15,8:17,*,*,*) 4.5.when Output还可以使用 when来实现达到某个固定条件再输出的效果，一般通过变 量，用户自定义的 函数以及 output内置的属性来实现。基本语法如下： plain view plaincopy 1. output after suppression_def 2. all | first | last | snapshot when trigger_expression 3. then set variable_name = assign_expression , variable_name = assign_expression ,. trigger_expression返回 true或者 false，表示输出或者不输出 then set variable_name=assign_expression表示是当 trigger_expression 被触发时，可对变量重新赋值。完整例子如下： java view plaincopy 1. /* 2. * 3. * author luonanqin 4. * 5. */ 6. class Pink 7. 8. private int id; 9. private int price; 10. 11. public int getId() 12. 13. return id; 14. 15. 16. public void setId(int id) 17. 18. this.id = id; 19. 20. 21. public int getPrice() 22. 23. return price; 24. 25. 26. public void setPrice(int price) 27. 28. this.price = price; 29. 30. 31. public String toString() 32. 33. return “id: “ + id + “, price: “ + price; 34. 35. 36. 37.class OutputWhenListener implements UpdateListener 38. 39. public void update(EventBean newEvents, EventBean oldEv ents) 40. 41. if (newEvents != null) 42. 43. for (int i = 0; i 5) 82. 83. runtime.setVariableValue(“exceed“, true); 84. / 因为主线程和输出线程不是同一个，所以这里休息 1秒保证输出线程将事件全部输出，方便演示。 85. Thread.sleep(1000); 86. 87. 88. 89. 执行结果： plain view plaincopy 1. Send Pink Event: id: 1, price: 8 2. Output Pink: id: 1, price: 8 3. Send Pink Event: id: 2, price: 5 4. Send Pink Event: id: 3, price: 3 5. Send Pink Event: id: 4, price: 1 6. Send Pink Event: id: 5, price: 1 7. Send Pink Event: id: 6, price: 9 8. Output Pink: id: 2, price: 5 9. Output Pink: id: 3, price: 3 10.Output Pink: id: 4, price: 1 11.Output Pink: id: 5, price: 1 12.Output Pink: id: 6, price: 9 13.Send Pink Event: id: 7, price: 8 14.Output Pink: id: 7, price: 8 15.Send Pink Event: id: 8, price: 0 16.Send Pink Event: id: 9, price: 2 17.Send Pink Event: id: 10, price: 7 18.Output Pink: id: 8, price: 0 19.Output Pink: id: 9, price: 2 20.Output Pink: id: 10, price: 7 从结果可以看出来。当 price大于 5的时候，设置 exceed变量为 true，即可输出之前进入的所有事件，then set 子句将 exceed设置为 false， 等待下一次 exceed=true时触发输出。由于输出线程是单独的线程，所以如果 不 sleep，结果可能会 和这个不同。 对于 when关键字，Esper 提供了一些内置的属性帮助我们实现更复杂的输出约 束。如图所示： 以上 5个属性我就不多做解释了，使用方式是作为 trigger_expression跟在 when关键字的后面。例如： plain view plaincopy 1. / 进入的 Apple事件总数达到 5个时才输出，且不清零 count_insert_total属性，继续累加事件总数 2. select * from Apple output when count_insert_total=5 3. 4. / 移除的 Apple事件总数达到 4个时才输出，并清零 count_remove属 性 5. select * from Apple output when count_remove=4 另外，在使用 when的时候，有两点需要注意： 1.当 trigger_expression返回 true时，Esper 会输出从上一次输出之后到这 次输出之间所有的 insert stream和