AI 助理
备案控制台
开发者社区 数据库 文章 正文

（5）Flink CEP SQL四种匹配模式效果演示

2022-08-24 439
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
实时计算 Flink 版，5000CU*H 3个月
推荐场景：
实时发现最热Github项目
简介： Flink CEP SQL中提供了四种匹配策略：（1）skip to next row从匹配成功的事件序列中的第一个事件的下一个事件开始进行下一次匹配（2）skip past last row从匹配成功的事件序列中的最后一个事件的下一个事件开始进行下一次匹配（3）skip to first pattern Item从匹配成功的事件序列中第一个对应于patternItem的事件开始进行下一次匹配（4）skip to last pattern Item从匹配成功的事件序列中最后一个对应于patternItem的事件开始进行下一次匹配

Flink CEP SQL中提供了四种匹配策略：
（1）skip to next row
从匹配成功的事件序列中的第一个事件的下一个事件开始进行下一次匹配
（2）skip past last row
从匹配成功的事件序列中的最后一个事件的下一个事件开始进行下一次匹配
（3）skip to first pattern Item
从匹配成功的事件序列中第一个对应于patternItem的事件开始进行下一次匹配
（4）skip to last pattern Item
从匹配成功的事件序列中最后一个对应于patternItem的事件开始进行下一次匹配

接下来我们代码来演示一下每种策略模式表达的效果：
（1）skip to next row

package com.examples;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * Created by lj on 2022-08-08.
 */
public class CEPSQLExampleAfterMatch {
    private static final Logger LOG = LoggerFactory.getLogger(CEPSQLExampleAfterMatch.class);

    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);
            final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            DataStream<Ticker> dataStream =
                    env.fromElements(
                            new Ticker(2, "Apple", 11, 2, LocalDateTime.parse("2021-12-10 10:00:01", dateTimeFormatter)),
                            new Ticker(3, "Apple", 16, 2, LocalDateTime.parse("2021-12-10 10:00:02", dateTimeFormatter)),
                            new Ticker(4, "Apple", 13, 2, LocalDateTime.parse("2021-12-10 10:00:03", dateTimeFormatter)),
                            new Ticker(5, "Apple", 15, 2, LocalDateTime.parse("2021-12-10 10:00:04", dateTimeFormatter)),
                            new Ticker(6, "Apple", 14, 1, LocalDateTime.parse("2021-12-10 10:00:05", dateTimeFormatter)),
                            new Ticker(7, "Apple", 19, 1, LocalDateTime.parse("2021-12-10 10:00:06", dateTimeFormatter)),
                            new Ticker(8, "Apple", 23, 2, LocalDateTime.parse("2021-12-10 10:00:07", dateTimeFormatter)),
                            new Ticker(9, "Apple", 22, 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_1", table);
            
            String sql = "SELECT * " +
                    "FROM CEP_SQL_1 " +
                    "    MATCH_RECOGNIZE ( " +
                    "        PARTITION BY symbol " +       //按symbol分区，将相同卡号的数据分到同一个计算节点上。
                    "        ORDER BY rowtime " +          //在窗口内，对事件时间进行排序。
                    "        MEASURES " +                   //定义如何根据匹配成功的输入事件构造输出事件
                    "            FIRST(e1.id) as id,"+
                    "            AVG(e1.price) as avgPrice,"+
                    "            FIRST(e1.rowtime) AS e1_start_tstamp, " +
                    "            LAST(e2.rowtime) AS e2_fast_tstamp " +
                    "        ONE ROW PER MATCH " +                                        //匹配成功输出一条
                    "        AFTER MATCH skip to next row " +
                    "        PATTERN ( e1+ e2) WITHIN INTERVAL '2' MINUTE" +
                    "        DEFINE " +                                                     //定义各事件的匹配条件
                    "            e1 AS " +
                    "                e1.price < 19 , " +
                    "            e2 AS " +
                    "                e2.price >= 19 " +
                    "    ) MR";
            
            
            TableResult res = tEnv.executeSql(sql);
            res.print();
            tEnv.dropTemporaryView("CEP_SQL_1");
                
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }
    }


    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;
        }
    }
}

image.png
（2）skip past last row

package com.examples;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * Created by lj on 2022-08-08.
 */
public class CEPSQLExampleAfterMatch {
    private static final Logger LOG = LoggerFactory.getLogger(CEPSQLExampleAfterMatch.class);

    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);
            final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            DataStream<Ticker> dataStream =
                    env.fromElements(
                            new Ticker(2, "Apple", 11, 2, LocalDateTime.parse("2021-12-10 10:00:01", dateTimeFormatter)),
                            new Ticker(3, "Apple", 16, 2, LocalDateTime.parse("2021-12-10 10:00:02", dateTimeFormatter)),
                            new Ticker(4, "Apple", 13, 2, LocalDateTime.parse("2021-12-10 10:00:03", dateTimeFormatter)),
                            new Ticker(5, "Apple", 15, 2, LocalDateTime.parse("2021-12-10 10:00:04", dateTimeFormatter)),
                            new Ticker(6, "Apple", 14, 1, LocalDateTime.parse("2021-12-10 10:00:05", dateTimeFormatter)),
                            new Ticker(7, "Apple", 19, 1, LocalDateTime.parse("2021-12-10 10:00:06", dateTimeFormatter)),
                            new Ticker(8, "Apple", 23, 2, LocalDateTime.parse("2021-12-10 10:00:07", dateTimeFormatter)),
                            new Ticker(9, "Apple", 22, 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_2", table);
            
            String sql = "SELECT * " +
                    "FROM CEP_SQL_2 " +
                    "    MATCH_RECOGNIZE ( " +
                    "        PARTITION BY symbol " +       //按symbol分区，将相同卡号的数据分到同一个计算节点上。
                    "        ORDER BY rowtime " +          //在窗口内，对事件时间进行排序。
                    "        MEASURES " +                   //定义如何根据匹配成功的输入事件构造输出事件
                    "            e1.id as id,"+
                    "            AVG(e1.price) as avgPrice,"+
                    "            FIRST(e1.rowtime) AS e1_start_tstamp, " +
                    "            LAST(e1.rowtime) AS e1_fast_tstamp, " +
                    "            e2.rowtime AS end_tstamp " +
                    "        ONE ROW PER MATCH " +                                        //匹配成功输出一条
                    "        AFTER MATCH skip past last row " +
                    "        PATTERN (e1+ e2) WITHIN INTERVAL '2' MINUTE" +
                    "        DEFINE " +                                                     //定义各事件的匹配条件
                    "            e1 AS " +
                    "                e1.price < 19 , " +
                    "            e2 AS " +
                    "                e2.price >= 19 " +
                    "    ) MR";
            
            
            TableResult res = tEnv.executeSql(sql);
            res.print();
            tEnv.dropTemporaryView("CEP_SQL_2");
                
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }
    }


    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;
        }
    }
}

image.png
（3）skip to first pattern Item

package com.examples;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * Created by lj on 2022-08-08.
 */
public class CEPSQLExampleAfterMatch {
    private static final Logger LOG = LoggerFactory.getLogger(CEPSQLExampleAfterMatch.class);

    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);
            final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            DataStream<Ticker> dataStream =
                    env.fromElements(
                            new Ticker(2, "Apple", 11, 2, LocalDateTime.parse("2021-12-10 10:00:01", dateTimeFormatter)),
                            new Ticker(3, "Apple", 16, 2, LocalDateTime.parse("2021-12-10 10:00:02", dateTimeFormatter)),
                            new Ticker(4, "Apple", 13, 2, LocalDateTime.parse("2021-12-10 10:00:03", dateTimeFormatter)),
                            new Ticker(5, "Apple", 15, 2, LocalDateTime.parse("2021-12-10 10:00:04", dateTimeFormatter)),
                            new Ticker(6, "Apple", 14, 1, LocalDateTime.parse("2021-12-10 10:00:05", dateTimeFormatter)),
                            new Ticker(7, "Apple", 19, 1, LocalDateTime.parse("2021-12-10 10:00:06", dateTimeFormatter)),
                            new Ticker(8, "Apple", 23, 2, LocalDateTime.parse("2021-12-10 10:00:07", dateTimeFormatter)),
                            new Ticker(9, "Apple", 22, 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_3", table);
            
            String sql = "SELECT * " +
                    "FROM CEP_SQL_3 " +
                    "    MATCH_RECOGNIZE ( " +
                    "        PARTITION BY symbol " +       //按symbol分区，将相同卡号的数据分到同一个计算节点上。
                    "        ORDER BY rowtime " +          //在窗口内，对事件时间进行排序。
                    "        MEASURES " +                   //定义如何根据匹配成功的输入事件构造输出事件
                    "            FIRST(e1.id) as id,"+
                    "            AVG(e1.price) as avgPrice,"+
                    "            FIRST(e1.rowtime) AS e1_start_tstamp, " +
                    "            LAST(e1.rowtime) AS e1_fast_tstamp, " +
                    "            e2.rowtime AS end_tstamp " +
                    "        ONE ROW PER MATCH " +                                        //匹配成功输出一条
                    "        AFTER MATCH skip to first e1 " +
                    "        PATTERN (e0 e1+ e2) WITHIN INTERVAL '2' MINUTE" +
                    "        DEFINE " +                                                     //定义各事件的匹配条件
                    "            e1 AS " +
                    "                e1.price < 19 , " +
                    "            e2 AS " +
                    "                e2.price >= 19 " +
                    "    ) MR";
            
            
            TableResult res = tEnv.executeSql(sql);
            res.print();
            tEnv.dropTemporaryView("CEP_SQL_3");
                
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }
    }


    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;
        }
    }
}

image.png
（4）skip to last pattern Item

package com.examples;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * Created by lj on 2022-08-08.
 */
public class CEPSQLExampleAfterMatch {
    private static final Logger LOG = LoggerFactory.getLogger(CEPSQLExampleAfterMatch.class);

    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);
            final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            DataStream<Ticker> dataStream =
                    env.fromElements(
                            new Ticker(2, "Apple", 11, 2, LocalDateTime.parse("2021-12-10 10:00:01", dateTimeFormatter)),
                            new Ticker(3, "Apple", 16, 2, LocalDateTime.parse("2021-12-10 10:00:02", dateTimeFormatter)),
                            new Ticker(4, "Apple", 13, 2, LocalDateTime.parse("2021-12-10 10:00:03", dateTimeFormatter)),
                            new Ticker(5, "Apple", 15, 2, LocalDateTime.parse("2021-12-10 10:00:04", dateTimeFormatter)),
                            new Ticker(6, "Apple", 14, 1, LocalDateTime.parse("2021-12-10 10:00:05", dateTimeFormatter)),
                            new Ticker(7, "Apple", 19, 1, LocalDateTime.parse("2021-12-10 10:00:06", dateTimeFormatter)),
                            new Ticker(8, "Apple", 23, 2, LocalDateTime.parse("2021-12-10 10:00:07", dateTimeFormatter)),
                            new Ticker(9, "Apple", 22, 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_4", table);
            
            String sql = "SELECT * " +
                    "FROM CEP_SQL_4 " +
                    "    MATCH_RECOGNIZE ( " +
                    "        PARTITION BY symbol " +       //按symbol分区，将相同卡号的数据分到同一个计算节点上。
                    "        ORDER BY rowtime " +          //在窗口内，对事件时间进行排序。
                    "        MEASURES " +                   //定义如何根据匹配成功的输入事件构造输出事件
                    "            e1.id as id,"+
                    "            AVG(e1.price) as avgPrice,"+
                    "            FIRST(e1.rowtime) AS e1_start_tstamp, " +
                    "            LAST(e1.rowtime) AS e1_fast_tstamp, " +
                    "            e2.rowtime AS end_tstamp " +
                    "        ONE ROW PER MATCH " +                                        //匹配成功输出一条
                    "        AFTER MATCH skip to last e1 " +
                    "        PATTERN (e0 e1+ e2) WITHIN INTERVAL '2' MINUTE" +
                    "        DEFINE " +                                                     //定义各事件的匹配条件
                    "            e1 AS " +
                    "                e1.price < 19 , " +
                    "            e2 AS " +
                    "                e2.price >= 19 " +
                    "    ) MR";
            
            
            TableResult res = tEnv.executeSql(sql);
            res.print();
            tEnv.dropTemporaryView("CEP_SQL_4");
                
        } catch (Exception e) {
            LOG.error(e.getMessage(), e);
        }
    }


    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;
        }
    }
}

image.png

文章标签：
实时计算 Flink版
流计算
SQL
关键词：
实时计算 Flink版模式
实时计算 Flink版CEP
flink SQL
实时计算 Flink版SQL
SQL flink
相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程，适合初学者阅读。由浅入深案例丰富，通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员，只要能够坚持把所有章节都学完，也一定会受益匪浅。
NBI大数据可视化
目录
相关文章
百岁001
|
14天前
|
关系型数据库 MySQL 数据库
基于Flink CDC 开发，支持Web-UI的实时KingBase 连接器，三大模式无缝切换，效率翻倍！
TIS 是一款基于Web-UI的开源大数据集成工具，通过与人大金仓Kingbase的深度整合，提供高效、灵活的实时数据集成方案。它支持增量数据监听和实时写入，兼容MySQL、PostgreSQL和Oracle模式，无需编写复杂脚本，操作简单直观，特别适合非专业开发人员使用。TIS率先实现了Kingbase CDC连接器的整合，成为业界首个开箱即用的Kingbase CDC数据同步解决方案，助力企业数字化转型。
百岁001
67 5
基于Flink CDC 开发，支持Web-UI的实时KingBase 连接器，三大模式无缝切换，效率翻倍！
技术小达人
|
24天前
|
消息中间件 JSON 数据库
探索Flink动态CEP：杭州银行的实战案例
探索Flink动态CEP：杭州银行的实战案例
技术小达人
36 5
申某某
|
2月前
|
SQL 大数据 数据处理
Flink SQL 详解：流批一体处理的强大工具
Flink SQL 是为应对传统数据处理框架中流批分离的问题而诞生的，它融合了SQL的简洁性和Flink的强大流批处理能力，降低了大数据处理门槛。其核心工作原理包括生成逻辑执行计划、查询优化和构建算子树，确保高效执行。Flink SQL 支持过滤、投影、聚合、连接和窗口等常用算子，实现了流批一体处理，极大提高了开发效率和代码复用性。通过统一的API和语法，Flink SQL 能够灵活应对实时和离线数据分析场景，为企业提供强大的数据处理能力。
申某某
308 26
武子康
|
5月前
|
分布式计算 监控 大数据
大数据-131 - Flink CEP 案例：检测交易活跃用户、超时未交付
大数据-131 - Flink CEP 案例：检测交易活跃用户、超时未交付
武子康
144 0
武子康
|
5月前
|
分布式计算 资源调度 大数据
大数据-110 Flink 安装部署 下载解压配置 Standalone模式启动 打包依赖（一）
大数据-110 Flink 安装部署 下载解压配置 Standalone模式启动 打包依赖（一）
武子康
171 0
灵杰开发者
|
3月前
|
消息中间件 JSON 数据库
探索Flink动态CEP：杭州银行的实战案例
本文由杭州银行大数据工程师唐占峰、欧阳武林撰写，介绍Flink动态CEP的定义、应用场景、技术实现及使用方式。Flink动态CEP是基于Flink的复杂事件处理库，支持在不重启服务的情况下动态更新规则，适应快速变化的业务需求。文章详细阐述了其在反洗钱、反欺诈和实时营销等金融领域的应用，并展示了某金融机构的实际应用案例。通过动态CEP，用户可以实时调整规则，提高系统的灵活性和响应速度，降低维护成本。文中还提供了具体的代码示例和技术细节，帮助读者理解和使用Flink动态CEP。
灵杰开发者
668 2
探索Flink动态CEP：杭州银行的实战案例
申某某
|
3月前
|
SQL 存储 缓存
Flink SQL Deduplication 去重以及如何获取最新状态操作
Flink SQL Deduplication 是一种高效的数据去重功能，支持多种数据类型和灵活的配置选项。它通过哈希表、时间窗口和状态管理等技术实现去重，适用于流处理和批处理场景。本文介绍了其特性、原理、实际案例及源码分析，帮助读者更好地理解和应用这一功能。
申某某
248 14
武子康
|
5月前
|
SQL 大数据 API
大数据-132 - Flink SQL 基本介绍 与 HelloWorld案例
大数据-132 - Flink SQL 基本介绍 与 HelloWorld案例
武子康
111 0
武子康
|
5月前
|
SQL 消息中间件 分布式计算
大数据-130 - Flink CEP 详解 - CEP开发流程 与 案例实践：恶意登录检测实现
大数据-130 - Flink CEP 详解 - CEP开发流程 与 案例实践：恶意登录检测实现
武子康
186 0
武子康
|
5月前
|
分布式计算 监控 大数据
大数据-129 - Flink CEP 详解 Complex Event Processing - 复杂事件处理
大数据-129 - Flink CEP 详解 Complex Event Processing - 复杂事件处理
武子康
122 0

热门文章

最新文章

  • 1
    Flink基于Paimon的实时湖仓解决方案的演进
  • 2
    阿里妈妈基于 Flink+Paimon 的 Lakehouse 应用实践
  • 3
    基于 Flink CDC YAML 的 MySQL 到 Kafka 流式数据集成
  • 4
    Flink CDC + Hologres高性能数据同步优化实践
  • 5
    基于 Flink CDC YAML 的 MySQL 到 Kafka 流式数据集成
  • 6
    基于Flink CDC 开发，支持Web-UI的实时KingBase 连接器，三大模式无缝切换，效率翻倍！
  • 7
    鹰角基于 Flink + Paimon + Trino 构建湖仓一体化平台实践项目
  • 8
    Flink CDC 在阿里云 DataWorks 数据集成入湖场景的应用实践
  • 9
    Flink CDC 3.3.0 发布公告
  • 10
    万字长文带你深入广告场景Paimon+Flink全链路探索与实践
  • 1
    从入门到精通：Python Web安全守护指南，SQL注入、XSS、CSRF全防御！
    104
  • 2
    逆袭黑客！掌握Python Web安全精髓，让SQL注入、XSS、CSRF无所遁形！
    229
  • 3
    WAF如何防御SQL注入？
    447
  • 4
    如何做好SQL质量监控
    56
  • 5
    现代深度学习框架构建问题之深度学习通用架构的定义如何解决
    75
  • 6
    面试题MySQL问题之实现覆盖索引如何解决
    85
  • 7
    ODPS SQL问题之为什么使用odps.sql.groupby.skewindata = true优化后，逻辑执行计划会发生改变如何解决
    176
  • 8
    ODPS SQL问题之在ODPS SQL中，Shuffle的概念是什么
    72
  • 9
    sql函数优化
    40
  • 10
    SQL分组函数
    62

    • 相关课程

    更多
  • 如何在 PolarDB-X 中优化慢 SQL
  • SQL完全自学手册
  • SQL Server on Linux入门教程
  • SQL入门与实践
  • 数据库及SQL/MySQL基础
  • SQL进阶及查询

    • 相关电子书

    更多
  • 低代码开发师（初级）实战教程
  • 冬季实战营第三期：MySQL数据库进阶实战
  • 阿里巴巴DevOps 最佳实践手册

    • 相关实验场景

    更多
  • 基于Hologres+Flink搭建GitHub实时数据大屏
  • PolarDB for AI：在数据库中通过SQL实现AI能力
  • 玩转MaxCompute SQL! 30分钟搞定数据分析挖掘
  • 基于Flink+Tair搭建实时监控大屏
  • 5分钟上手Flink MySQL连接器
  • 使用Flink实时发现最热Github项目
    • 下一篇
    阿里云oss简介和如何对接使用