上一篇我们对Flink CEP做了简单介绍，这一篇我们通过代码来演示一下Flink CEP SQL中的严格近邻效果：
（1）pom依赖：
代码语言：javascript复制<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-cep_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>（2）定义一个消息对象
代码语言：javascript复制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）构造数据，定义事件组合
代码语言：javascript复制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）执行效果：
从数据集中匹配到了两组符合要求的数据。
