上一篇咱们对Flink CEP做了简略介绍，这一篇咱们通过代码来演示一下Flink CEP SQL中的严格近邻成果：
（1）pom依赖：<dependency>
<groupId>org.apache.flink</groupId><artifactId>flink-cep_${scala.binary.version}</artifactId><version>${flink.version}</version></dependency>（2）定义一个音讯对象public static class Ticker {
public long id;public String symbol;public long price;public long tax;public LocalDateTime rowtime;public Ticker() {}public Ticker(long id, String symbol, long price, long item, LocalDateTime rowtime) { this.id = id; this.symbol = symbol; this.price = price; this.tax = tax; this.rowtime = rowtime;}}（3）结构数据，定义事件组合public static void main(String[] args) {
EnvironmentSettings settings = null;StreamTableEnvironment tEnv = null;try { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); settings = EnvironmentSettings.newInstance() .useBlinkPlanner() .inStreamingMode() .build(); tEnv = StreamTableEnvironment.create(env, settings); System.out.println("===============CEP_SQL_9================="); final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); DataStream<Ticker> dataStream = env.fromElements( new Ticker(1, "ACME", 22, 1, LocalDateTime.parse("2021-12-10 10:00:00", dateTimeFormatter)), new Ticker(3, "ACME", 19, 1, LocalDateTime.parse("2021-12-10 10:00:02", dateTimeFormatter)), new Ticker(4, "ACME", 23, 3, LocalDateTime.parse("2021-12-10 10:00:03", dateTimeFormatter)), new Ticker(5, "Apple", 25, 2, LocalDateTime.parse("2021-12-10 10:00:04", dateTimeFormatter)), new Ticker(6, "Apple", 18, 1, LocalDateTime.parse("2021-12-10 10:00:05", dateTimeFormatter)), new Ticker(7, "Apple", 16, 1, LocalDateTime.parse("2021-12-10 10:00:06", dateTimeFormatter)), new Ticker(8, "Apple", 14, 2, LocalDateTime.parse("2021-12-10 10:00:07", dateTimeFormatter)), new Ticker(9, "Apple", 15, 2, LocalDateTime.parse("2021-12-10 10:00:08", dateTimeFormatter)), new Ticker(10, "Apple", 25, 2, LocalDateTime.parse("2021-12-10 10:00:09", dateTimeFormatter)), new Ticker(11, "Apple", 22, 1, LocalDateTime.parse("2021-12-10 10:00:11", dateTimeFormatter)), new Ticker(12, "Apple", 15, 1, LocalDateTime.parse("2021-12-10 10:00:12", dateTimeFormatter)), new Ticker(13, "Apple", 19, 1, LocalDateTime.parse("2021-12-10 10:00:13", dateTimeFormatter)), new Ticker(14, "Apple", 25, 1, LocalDateTime.parse("2021-12-10 10:00:14", dateTimeFormatter)), new Ticker(15, "Apple", 19, 1, LocalDateTime.parse("2021-12-10 10:00:15", dateTimeFormatter)), new Ticker(16, "Apple", 15, 1, LocalDateTime.parse("2021-12-10 10:00:16", dateTimeFormatter)), new Ticker(17, "Apple", 19, 1, LocalDateTime.parse("2021-12-10 10:00:17", dateTimeFormatter)), new Ticker(18, "Apple", 15, 1, LocalDateTime.parse("2021-12-10 10:00:18", dateTimeFormatter))); Table table = tEnv.fromDataStream(dataStream, Schema.newBuilder() .column("id", DataTypes.BIGINT()) .column("symbol", DataTypes.STRING()) .column("price", DataTypes.BIGINT()) .column("tax", DataTypes.BIGINT()) .column("rowtime", DataTypes.TIMESTAMP(3)) .watermark("rowtime", "rowtime - INTERVAL '1' SECOND") .build()); tEnv.createTemporaryView("CEP_SQL_9", table); String sql = "SELECT * " + "FROM CEP_SQL_9 " + " MATCH_RECOGNIZE ( " + " PARTITION BY symbol " + //按symbol分区，将雷同卡号的数据分到同一个计算节点上。 " ORDER BY rowtime " + //在窗口内，对事件工夫进行排序。 " MEASURES " + //定义如何依据匹配胜利的输出事件结构输入事件 " e1.id as id,"+ " AVG(e1.price) as avgPrice,"+ " e1.rowtime AS start_tstamp, " + " e3.rowtime AS end_tstamp " + " ONE ROW PER MATCH " + //匹配胜利输入一条 " AFTER MATCH skip to next row " + //匹配后跳转到下一行 " PATTERN ( e1 e2 e3) WITHIN INTERVAL '2' MINUTE" + " DEFINE " + //定义各事件的匹配条件 " e1 AS " + " e1.price = 25 , " + " e2 AS " + " e2.price > 10 ," + " e3 AS " + " e3.price = 15 " + " ) MR"; TableResult res = tEnv.executeSql(sql); res.print(); tEnv.dropTemporaryView("CEP_SQL_9"); } catch (Exception e) { LOG.error(e.getMessage(), e); }}（4）要害代码解释：
输入两分钟内匹配到的数据，输入信息：
（5）执行成果：
从数据集中匹配到了两组符合要求的数据。