在之前的文章中介绍了FlinkCEP的基本使用方法，本文将介绍flink提供的sql方式实现模式匹配，即Detecting Patterns in Tables.

完整样例

代码传送门

  1. import org.apache.flink.api.common.typeinfo.TypeInformation;
  2. import org.apache.flink.api.common.typeinfo.Types;
  3. import org.apache.flink.formats.csv.CsvRowDeserializationSchema;
  4. import org.apache.flink.streaming.api.TimeCharacteristic;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
  8. import org.apache.flink.streaming.api.watermark.Watermark;
  9. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
  10. import org.apache.flink.table.api.Table;
  11. import org.apache.flink.table.api.TableSchema;
  12. import org.apache.flink.table.api.java.StreamTableEnvironment;
  13. import org.apache.flink.types.Row;
  14. import java.util.Properties;
  16. public class FlinkCEPSqlExample {
  17.     public static void main(String[] args) throws Exception {
  18.         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  20.         env.setParallelism(1);
  21.         final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  22.         final TableSchema tableSchema = new TableSchema(new String[]{"symbol","tax","price", "rowtime"}, new TypeInformation[]{Types.STRING, Types.STRING, Types.LONG, Types.SQL_TIMESTAMP});
  23.         final TypeInformation<Row> typeInfo = tableSchema.toRowType();
  24.         final CsvRowDeserializationSchema.Builder deserSchemaBuilder = new CsvRowDeserializationSchema.Builder(typeInfo).setFieldDelimiter(',');
  26.         Properties properties = new Properties();
  27.         properties.setProperty("bootstrap.servers", "host-10-1-236-139:6667");
  29.         FlinkKafkaConsumer010<Row> myConsumer = new FlinkKafkaConsumer010<>(
  30.                 "foo",
  31.                 deserSchemaBuilder.build(),
  32.                 properties);
  34.         myConsumer.setStartFromLatest();
  36.         DataStream<Row> stream = env.addSource(myConsumer).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator());
  37.         tableEnv.registerDataStream("Ticker", stream, "symbol,tax,price,rowtime.rowtime");
  39.         Table result = tableEnv.sqlQuery("SELECT * " +
  40.                 "FROM Ticker " +
  41.                 "    MATCH_RECOGNIZE( " +
  42.                 "        PARTITION BY symbol " +
  43.                 "        ORDER BY rowtime " +
  44.                 "        MEASURES " +
  45.                 "            A.price AS firstPrice, " +
  46.                 "            B.price AS lastPrice " +
  47.                 "        ONE ROW PER MATCH " +
  48.                 "        AFTER MATCH SKIP PAST LAST ROW " +
  49.                 "        PATTERN (A+ B) " +
  50.                 "        DEFINE " +
  51.                 "            A AS A.price < 10, " +
  52.                 "            B AS B.price > 100 " +
  53.                 "    )");
  55.         final TableSchema tableSchemaResult = new TableSchema(new String[]{"symbol","firstPrice","lastPrice"}, new TypeInformation[]{Types.STRING, Types.LONG, Types.LONG});
  56.         final TypeInformation<Row> typeInfoResult = tableSchemaResult.toRowType();
  57.         DataStream ds = tableEnv.toAppendStream(result, typeInfoResult);
  58.         ds.print();
  59.         env.execute("Flink CEP via SQL example");
  60.     }
  62.     private static class BoundedOutOfOrdernessGenerator implements AssignerWithPeriodicWatermarks<Row> {
  63.         private final long maxOutOfOrderness = 5000;
  64.         private long currentMaxTimestamp;
  66.         @Override
  67.         public long extractTimestamp(Row row, long previousElementTimestamp) {
  68.             System.out.println("Row is " + row);
  69.             long timestamp = StringUtilsPlus.dateToStamp(String.valueOf(row.getField(3)));
  70.             currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
  71.             System.out.println("watermark:" + StringUtilsPlus.stampToDate(String.valueOf(currentMaxTimestamp - maxOutOfOrderness)));
  72.             return timestamp;
  73.         }
  75.         @Override
  76.         public Watermark getCurrentWatermark() {
  77.             return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
  78.         }
  79.     }
  80. }

样例说明

  1. PARTITION BY symbol根据symbol字段进行逻辑分区
  2. ORDER BY rowtime根据事件时间进行排序，避免时序混乱
  3. MEASURES里面定义了要输出的字段
  4. ONE ROW PER MATCH为输出模式，另一种模式为ALL ROWS PER MATCH，但是目前flink1.9中只支持ONE ROW PER MATCH
  5. AFTER MATCH SKIP PAST LAST ROW匹配后的事件丢弃策略，这种策略保证每个事件最多匹配一次
  • SKIP PAST LAST ROW - resumes the pattern matching at the next row after the last row of the current match.
  • SKIP TO NEXT ROW - continues searching for a new match starting at the next row after the starting row of the match.
  • SKIP TO LAST variable - resumes the pattern matching at the last row that is mapped to the specified pattern variable.
  • SKIP TO FIRST variable - resumes the pattern matching at the first row that is mapped to the specified pattern variable.
  1. PATTERN (A+ B)定义了两个模式的关系，如果是(A B)表示两个事件必须第一个事件满足A的条件，第二个事件满足B的条件。(A+ B)的含义就是在满足B的条件之前，可以有1个或多个事件满足A的条件。具体规则参见Defining a Pattern
  2. DEFINE中定义了具体的每个模式的规则
  3. 虽然是以sql的形式运行，但是最终内部的运行逻辑和《Flink CEP进阶》中介绍的是一致的，只是flink对SQL进行了解析

总结

本文提供了一个简单的FlinkCEP SQL样例，可以让读者快速体验FlinkCEP SQL的运行效果，本例的业务含义可以理解为，连续小额交易后突然有一笔大额交易即输出结果。