关于flink:2Flink-CEP-SQL严格近邻代码演示风控系统构建利器

上一篇咱们对 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）执行成果：

从数据集中匹配到了两组符合要求的数据。