基于上一篇（3）Flink CEP SQL寬松近鄰代碼演示的延展，在上一篇中我們使用貪婪詞量 +（至少匹配1行或多行），本篇將演示多種貪婪詞量的效果：
（1）使用貪婪詞量 *（匹配0行或多行）
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_10=================");
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", 19, 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", 11, 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_10", table);
String sql = "SELECT * " +
"FROM CEP_SQL_10 " +
" MATCH_RECOGNIZE ( " +
" PARTITION BY symbol " + //按symbol分區(qū)，將相同卡號(hào)的數(shù)據(jù)分到同一個(gè)計(jì)算節(jié)點(diǎn)上。
" ORDER BY rowtime " + //在窗口內(nèi)，對(duì)事件時(shí)間進(jìn)行排序。
" MEASURES " + //定義如何根據(jù)匹配成功的輸入事件構(gòu)造輸出事件
" 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 " + //匹配后跳轉(zhuǎn)到下一行
" PATTERN ( e1 e2* e3) WITHIN INTERVAL '2' MINUTE" +
" DEFINE " + //定義各事件的匹配條件
" e1 AS " +
" e1.price = 25 , " +
" e2 AS " +
" e2.price > 10 AND e2.price <19," +
" e3 AS " +
" e3.price = 19 " +
" ) MR";
TableResult res = tEnv.executeSql(sql);
res.print();
tEnv.dropTemporaryView("CEP_SQL_10");
}
匹配到了三組數(shù)據(jù)
image.png
貪婪詞量 *（匹配0行或多行）
image.png
（2）使用貪婪詞量 {n}（嚴(yán)格匹配n行）
image.png
image.png
image.png
（3）使用貪婪詞量 {n，}（n或者更多行(n≥O)）
image.png
image.png
image.png
