Esper学习之九：EPL语法（五）

时间：2021-05-11 22:16:12

本篇的内容主要包括了Subquery（也就是子查询）和Join，内容不少，但是不难，基本上和sql差不太多。

1.Subquery

EPL里的Subquery和sql的类似，是否比sql的用法更多我不得而知，毕竟本人是sql菜鸟，只在where语句里用过子查询。废话不多说，先上几个Subquer的简单用法：

子查询结果作为外部事件的属性

  1. select assetId, (select zone from ZoneClosed.std:lastevent()) as lastClosed from RFIDEvent

上面的例子是说返回当前RFIDEvent的assetId属性值和最新ZoneClosed事件的zone属性值，且以lastClosed作为zone的别名。

子查询关联外部事件的属性

  1. select * from RfidEvent as RFID where 'Dock 1' = (select name from Zones.std:unique(zoneId) where zoneId = RFID.zoneId)

子查询语句中的where条件可以应用RFID的属性，即内部的zoneId=RFID.zoneId

  1. select zoneId, (select name from Zones.std:unique(zoneId) where zoneId = RFID.zoneId) as name from RFIDEvent

关联外部事件属性的同时也可以作为外部事件的属性返回。

子查询内部事件作为外部事件的属性

  1. select (select * from MarketData.std:lastevent()) as md from SuperMarket

每进入一个SuperMarket事件就返回最新的MarketData事件作为属性返回，别名为md

子查询中应用聚合函数

  1. select * from MarketData where price > (select max(price) from MarketData(symbol='GOOG').std:lastevent())

子查询出得最大price和当前事件的price进行比较

  1. select * from OrderEvent oe where qty > (select sum(qty) from OrderEvent.win:time(1 hour) pd where pd.client = oe.client)

子查询得出qty的总和和当前事件的qty进行比较

Filter中使用子查询

  1. select * from BarData(ticker='MSFT', closePrice < (select movAgv from SMA20Stream(ticker='MSFT').std:lastevent()))

子查询返回的movAgv和外部事件的属性closePrice进行比较作为外部事件的一个filter

Pattern中使用子查询

  1. select * from pattern [
  2. a=A -> b=B(bvalue = (select d_val from DNamedWindow as d where d.d_id = b.b_id and d.d_id = a.a_id))
  3. ]

pattern的含义可先不深究，这里只要知道子查询可以用在pattern中就行了。

Expression中使用子查询（什么是Expression？请看《Esper学习之五：EPL语法（一）》的第八点）

  1. expression subq {
  2. (select max(quantity) as maxq, min(quantity) as minq from OrderEvent.win:time(1 min))
  3. }
  4. select (quantity - minq) / (subq().maxq  - subq().minq) as prorated from OrderEvent


以上就是子查询的几种简单用法，不过有几点注意事项是要各位悉知的：


1.子查询的返回必须使用data window或者view来进行限制，控制子查询使用的事件数（data window和view相当于具有某种功能性的事件集合）


2.子查询语句只能由select子句，from子句以及where子句组成，其他的均不支持，比如group by，limit等


3.没有关联外部事件的子查询语句也可以使用聚合函数


4.子查询语句中的select子句必须对所有属性使用聚合函数


5.在使用子查询时，如果子查询的事件和外部事件类型一样，则事件到来时，先经过子查询语句的处理，然后再经过外部语句的处理。如果包含了多个子查询语句，则事件的处理顺序规则较为复杂，本人暂时没做研究。


针对第4点可能说得有些不明白，特此举例说明下：




    

  1. class Apple {
    2. 

  2. private int price;
    3. 

  3. private int size;
    4. 

  4. public void setPrice(int price) {
    5. 

  5. this.price = price;
    6. 

  6. }
    7. 

  7. public void setSize(int size) {
    8. 

  8. this.size = size;
    9. 

  9. }
    10. 

  10. public int getPrice() {
    11. 

  11. return price;
    12. 

  12. }
    13. 

  13. public int getSize() {
    14. 

  14. return size;
    15. 

  15. }
    16. 

  16. }
    17. 

  17. class Fruit {
    18. 

  18. }
    19. 

  19. public class Test {
    20. 

  20. public static void main(String[] args) throws InterruptedException {
    21. 

  21. EPServiceProvider epService = EPServiceProviderManager.getDefaultProvider();
    22. 

  22. EPAdministrator admin = epService.getEPAdministrator();
    23. 

  23. String epl1 = "select (select sum(price), sum(size) from " + Apple.class.getName() + ".std:lastevent()) from " + Fruit.class.getName();
    24. 

  24. /*
    25. 

  25. * 当然，对不同的属性使用不同的聚合函数也是可以的
    26. 

  26. */
    27. 

  27. // String epl1 = "select (select sum(price), avg(size) from " + Apple.class.getName() + ".std:lastevent()) from " + Fruit.class.getName();
    28. 

  28. /*
    29. 

  29. * 注意：size没有使用聚合函数，会导致创建epl失败。文档中注明了“The properties of the subselect stream must all be within aggregation functions”.
    30. 

  30. * 即子查询中的select子句使用聚合函数时，所查询的属性都要使用聚合函数
    31. 

  31. */
    32. 

  32. // String epl1 = "select (select sum(price), size from " + Apple.class.getName() + ".std:lastevent()) from " + Fruit.class.getName();
    33. 

  33. admin.createEPL(epl1);
    34. 

  34. System.out.println("Create epl successfully!");
    35. 

  35. }
    36. 

  36. }
    37. 





除了上面的语法外，有几个关键字也是需要大家注意的。


exists


这个exists和exist可不一样，虽说也是用在where子句中（不仅仅是where子句），但是exists前面是没有什么待比较属性的。语法如下：




  1. exists (subquery)




如果exists后面的subquery有查询结果，则返回true，如果没有查询结果就返回false。所以用在where子句中的示例如下：




  1. select * from Fruit as F where exists (select * from Apple.std:lastevent() where acount = F.fcount)




既然exists语法返回的时布尔值，那么可不可以用工在Filter中呢？经本人测试，确实可以用。示例如下：




  1. select * from Fruit(exists (select * from Apple.win:keepall()))




不过在使用exists有两个注意点：


1.exists后面的子查询语句一定要用圆括号括起来，遗漏的话会报语法错误。


2.exists后面的子查询语句只能返回单列值。什么叫单列值？比如：Apple有price和size属性，那么select price和select size都是返回的单列（即一列），select price, size就是多列（即两列）。如果subquery中返回的大于一列，则会报multi-column错误。谨记！


in / not in


用法和sql的一样，具体语法就不说了，直接看例子吧：




    

  1. // 当前进入的Apple事件的aPrice存在于过去十秒内进入的Fruit的fPrice即可返回
    2. 

  2. select * from Apple where aPrice in (select fPrice from Fruit.win:time(10 s))
    3. 





注意：从语法上说in/not in前面是expression，并没有规定只能是属性。比如：Apple有aPrice和aSize两个int属性，则where子句可以写成"aPrice+aSize in ....."（这里只是举例，不expression不表示任何意思）


any / some / all


除了用in或者not in判断属性值是否存在于子查询结果中，还可以使用any/some/all并配合一些比较符号与子查询结果进行比较。语法如下：




    

  1. expression operator any/some/all (subquery)
    2. 

  2. // operator包含：>=, !=, <>, <, <=, >, >=
    3. 





一个简单的例子：




    

  1. // 如果当前Apple事件的aPrice小于前十个Fruit中的任何一个fPrice，则返回此事件
    2. 

  2. select * from Apple where aPrice < any (select fPrice from Fruit.win:length(10))
    3. 

  3. // 如果当前Apple事件的aPrice小于前十个Fruit中的所有fPrice，则返回此事件
    4. 

  4. select * from Apple where aPrice < all (select fPrice from Fruit.win:length(10))
    5. 





some和any同义，所以用法也相同。既然一样为什么又要弄这么个关键字，我也没搞懂。。。


子查询返回多列数据


子查询可以返回单列数据，也可以返回多列数据。用法很简单，只需要“."（点）就能找到每列的数据了。示例如下：




    

  1. select *,
    2. 

  2. (select bid, offer from MarketData.std:unique(symbol) as md where md.symbol = oe.symbol) as bidoffer
    3. 

  3. from OrderEvent oe
    4. 





如上所示，子查询语句中包含bid和offer两列，并且这个语句的结果用bidoffer作为别名，所以要想得到bid和offer的具体数据只需要bidoffer.bid和bidoffer.offer即可，简单吧。子查询返回多行数据和返回多列数据类似，只不过数据会以数组形式是返回。


2.Join


Join在sql里是很常见的查询方法，EPL也同样支持，并且包含了full outer join / left outer join / right outer join / inner join等。和sql基本无差别。


inner join


在没有任何关键字的修饰下，即为默认join方式，也就是inner join。必须等到所有join的事件都到了才可能输出，因为要是有where关联两个事件，得满足where条件了才能输出。例如：




  1. select * from Orange.std:lastevent(), Banana.std:lastevent()




如果只有Orange到或者Banana到都不会输出内容。std:lastevent是一种data window。如果不加特殊修饰的话（特殊修饰？下面会告诉你），事件必须有data window或者view修饰，否则会出现语法错误。当然，不同的事件可以用不同的data window修饰，并没有规定要一样。


评论里c77_cn有问到：


为什么"select a.id, b.field from Event as a, method : Static Class . getMethod (a.id) as b"没有window或者view也报错？


后来我重读了一下官方文档，没有指明必须要window或者view。我这么写是因为我测时候的时候，错误里说明了需要window或者view（在没有unidirectional修饰的情况下，这个修饰就是刚才说的“特殊修饰”）。这个句子正确我猜测method方法是每到一个事件都会被调用，所以能够完成join的工作。如果不用method，在没有window或者view的情况下，两类不同的事件分先后进入引擎，那先进入的的事件不能暂存的话如何完成join呢？


full outer join


上面看到的默认join方式是要求所有join的事件都必须到达引擎才会输出，并且join的事件之间通过where子句设置了条件判断，还得到达的两个事件满足条件了才能输出，而full outer join正好解决了这个问题，不管哪个事件到达，不管是否符合条件，都会输出。例如：




  1. select * from Orange.std:lastevent() as o full outer join Banana.std:lastevent() as b on o.price = b.price




输出结果有4种可能：


a.当只有Orange事件到达，没有满足join条件，会输出Orange事件，且Banana事件为null。


b.当只有Banana事件到达，没有满足join条件，会输出Banana事件，且Orange事件为null。


c.当两个事件都到达了，且没有满足join条件，即price不相等，则a，b情况各出现一次。


d.当两个事件都到达了，且满足join条件，即price相等，即可输出满足条件的事件。


所以说不管什么情况下，当前进入的事件都会输出，至于join的那个事件，满足即输出事件，不满足即输出null。


left outer join


full outer join输出了所进入的所有事件，不满足join条件的就输出null，而left outer join则规定关键字左边的事件可以即刻输出，而关键字右边的事件必须满足join条件才可输出。示例如下：




  1. select * from Pink.std:lastevent() as pi left outer join Pear.std:lastevent() as pe on pi.price = pe.price




因为Pink事件在left outer join的左边，所以他的输出不受join条件的限制，即事件到来该怎么输出怎么输出。但是Pear就不同，由于有join条件限制，即两个事件的price要相等，所以如果Pear事件到达的时候，如果没有满足条件的Pink事件，则Pear事件是不会输出的。（注意：输出null也算输出，这里是null都不会输出，即不触发listener）


right outer join


和left outer join相反，在关键字右边的事件不受join条件约束，而左边的事件必须满足join条件才可输出。具体例子我就不举了，大家可以写两个句子试试。


此外，在使用以上4种join的时候，可以多种join混用。比如：




    

  1. select * from Apple.std:lastevent() as a
    2. 

  2. left outer join Banana.std:lastevent() as b on a.price = b.price
    3. 

  3. full outer join Orange.std:lastevent() as o on o.price = a.price
    4. 





on后面的表达式是join的限制条件，且只能用“=”，如果想用其他操作符，则必须放到where子句中，这点尤其要注意。多个限制条件只能用and连接，不能用逗号，且限制的事件也要一样。比如：




    

  1. // a，b分别是两个事件的别名
    2. 

  2. // 正确写法
    3. 

  3. ……on a.price = b.price and a.size = b.size……
    4. 

  4. // 错误写法1：不能用逗号连接
    5. 

  5. ……on a.price = b.price, a.size = b.size……
    6. 

  6. // 错误写法2：必须针对同样的事件进行限制（c是另一个事件的别名）
    7. 

  7. ……on a.price = b.price and a.size = c.size……
    8. 





Unidirectional Join


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




  1. select * from Apple as a unidirectional, Banana.std:lastevent() as b where a.price = b.price




上句的意思是：维持最新的Banana事件，直到一个和Banana的price相等的Apple事件到来时输出两者。


由于有unidirectional的修饰，表明Apple事件是即时出发join操作，也就是说进入此EPL的Apple事件是无状态的。所以当Apple事件到来时，如果没有price相等的Banana，则什么输出也没有，即使下一个Banana事件的price和之前来的Apple的price相等也不会有输出，因为那个Apple事件已经从这个句子的上下文中移除了。为了更好的看到效果，我给大家写了一给完整的例子：




    

  1. class Orange {
    2. 

  2. private int price;
    3. 

  3. public void setPrice(int price) {
    4. 

  4. this.price = price;
    5. 

  5. }
    6. 

  6. public int getPrice() {
    7. 

  7. return price;
    8. 

  8. }
    9. 

  9. @Override
    10. 

  10. public String toString() {
    11. 

  11. return "Orange price=" + price;
    12. 

  12. }
    13. 

  13. }
    14. 

  14. class Banana {
    15. 

  15. private int price;
    16. 

  16. public int getPrice() {
    17. 

  17. return price;
    18. 

  18. }
    19. 

  19. public void setPrice(int price) {
    20. 

  20. this.price = price;
    21. 

  21. }
    22. 

  22. @Override
    23. 

  23. public String toString() {
    24. 

  24. return "Banana price=" + price;
    25. 

  25. }
    26. 

  26. }
    27. 

  27. class JoinUnidirectionalListener implements UpdateListener {
    28. 

  28. public void update(EventBean[] newEvents, EventBean[] oldEvents) {
    29. 

  29. if (newEvents != null) {
    30. 

  30. System.out.println(newEvents[0].get("o") + ", " + newEvents[0].get("b"));
    31. 

  31. }
    32. 

  32. }
    33. 

  33. }
    34. 

  34. public class JoinUnidirectionalTest {
    35. 

  35. public static void main(String[] args) throws InterruptedException {
    36. 

  36. EPServiceProvider epService = EPServiceProviderManager.getDefaultProvider();
    37. 

  37. EPAdministrator admin = epService.getEPAdministrator();
    38. 

  38. String epl1 = "select * from " + Orange.class.getName() + " as o unidirectional, " + Banana.class.getName()
    39. 

  39. + ".std:lastevent() as b where o.price = b.price";
    40. 

  40. EPStatement stat = admin.createEPL(epl1);
    41. 

  41. stat.addListener(new JoinUnidirectionalListener());
    42. 

  42. EPRuntime runtime = epService.getEPRuntime();
    43. 

  43. Orange o1 = new Orange();
    44. 

  44. o1.setPrice(1);
    45. 

  45. System.out.println("Send Orange1");
    46. 

  46. runtime.sendEvent(o1);
    47. 

  47. Banana b1 = new Banana();
    48. 

  48. b1.setPrice(1);
    49. 

  49. System.out.println("Send Banana1");
    50. 

  50. runtime.sendEvent(b1);
    51. 

  51. Banana b2 = new Banana();
    52. 

  52. b2.setPrice(2);
    53. 

  53. System.out.println("Send Banana2");
    54. 

  54. runtime.sendEvent(b2);
    55. 

  55. Orange o2 = new Orange();
    56. 

  56. o2.setPrice(2);
    57. 

  57. System.out.println("Send Orange2");
    58. 

  58. runtime.sendEvent(o2);
    59. 

  59. }
    60. 

  60. }
    61. 





执行结果：




    

  1. Send Orange1
    2. 

  2. Send Banana1
    3. 

  3. Send Banana2
    4. 

  4. Send Orange2
    5. 

  5. Orange price=2, Banana price=2
    6. 





可以看到o1和b1的price相等，o2和b2的price相等。o1先于b1进入引擎，由于o1进入时发现没有满足条件的Banana事件，所以什么输出也没有。之后b1进入了，因为满足条件的o1已经移除了，所以也是没有输出。b2先于o2进入引擎，被引擎暂存了起来，然后o2进入后，立刻进行join条件判断，发现暂存的b2的price相等，所以触发了listener并输出满足条件的这两个对象。


unidirectional使用很简单，但是也有其限制：


1.在一个join句子中，unidirectional关键字只能用于一个事件流。


2.用unidirectional修饰的事件流，不能通过esper的查询api查出来，因为该事件流是无状态的，不会暂存在引擎中，所以就没法查了。（关于查询api，后面的章节会详说）


3.使用了unidirectional修饰的事件流就不能再用data window或者view修饰了，也就是说他们是互斥的。
相关文章