(4)Flink CEP SQL贪婪词量演示

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 基于上一篇(3)Flink CEP SQL宽松近邻代码演示的延展,在上一篇中我们使用贪婪词量 +(至少匹配1行或多行),本篇将演示多种贪婪词量的效果:(1)使用贪婪词量 *(匹配0行或多行)

基于上一篇(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分区,将相同卡号的数据分到同一个计算节点上。
                "        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 AND e2.price <19," +
                "            e3 AS " +
                "                e3.price = 19 " +
                "    ) MR";
        
        
        TableResult res = tEnv.executeSql(sql);
        res.print();
        tEnv.dropTemporaryView("CEP_SQL_10");
}

匹配到了三组数据
image.png
贪婪词量 *(匹配0行或多行)
image.png
(2)使用贪婪词量 {n}(严格匹配n行)
image.png
image.png
image.png
(3)使用贪婪词量 {n,}(n或者更多行(n≥O))
image.png
image.png
image.png

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
11天前
|
SQL 大数据 数据处理
Flink SQL 详解:流批一体处理的强大工具
Flink SQL 是为应对传统数据处理框架中流批分离的问题而诞生的,它融合了SQL的简洁性和Flink的强大流批处理能力,降低了大数据处理门槛。其核心工作原理包括生成逻辑执行计划、查询优化和构建算子树,确保高效执行。Flink SQL 支持过滤、投影、聚合、连接和窗口等常用算子,实现了流批一体处理,极大提高了开发效率和代码复用性。通过统一的API和语法,Flink SQL 能够灵活应对实时和离线数据分析场景,为企业提供强大的数据处理能力。
106 26
|
5月前
|
SQL 存储 API
Flink实践:通过Flink SQL进行SFTP文件的读写操作
虽然 Apache Flink 与 SFTP 之间的直接交互存在一定的限制,但通过一些创造性的方法和技术,我们仍然可以有效地实现对 SFTP 文件的读写操作。这既展现了 Flink 在处理复杂数据场景中的强大能力,也体现了软件工程中常见的问题解决思路——即通过现有工具和一定的间接方法来克服技术障碍。通过这种方式,Flink SQL 成为了处理各种数据源,包括 SFTP 文件,在内的强大工具。
230 15
|
4月前
|
分布式计算 监控 大数据
大数据-131 - Flink CEP 案例:检测交易活跃用户、超时未交付
大数据-131 - Flink CEP 案例:检测交易活跃用户、超时未交付
108 0
|
2月前
|
消息中间件 JSON 数据库
探索Flink动态CEP:杭州银行的实战案例
本文由杭州银行大数据工程师唐占峰、欧阳武林撰写,介绍Flink动态CEP的定义、应用场景、技术实现及使用方式。Flink动态CEP是基于Flink的复杂事件处理库,支持在不重启服务的情况下动态更新规则,适应快速变化的业务需求。文章详细阐述了其在反洗钱、反欺诈和实时营销等金融领域的应用,并展示了某金融机构的实际应用案例。通过动态CEP,用户可以实时调整规则,提高系统的灵活性和响应速度,降低维护成本。文中还提供了具体的代码示例和技术细节,帮助读者理解和使用Flink动态CEP。
448 2
探索Flink动态CEP:杭州银行的实战案例
|
2月前
|
SQL 存储 缓存
Flink SQL Deduplication 去重以及如何获取最新状态操作
Flink SQL Deduplication 是一种高效的数据去重功能,支持多种数据类型和灵活的配置选项。它通过哈希表、时间窗口和状态管理等技术实现去重,适用于流处理和批处理场景。本文介绍了其特性、原理、实际案例及源码分析,帮助读者更好地理解和应用这一功能。
153 14
|
4月前
|
SQL 大数据 API
大数据-132 - Flink SQL 基本介绍 与 HelloWorld案例
大数据-132 - Flink SQL 基本介绍 与 HelloWorld案例
75 0
|
4月前
|
SQL 消息中间件 分布式计算
大数据-130 - Flink CEP 详解 - CEP开发流程 与 案例实践:恶意登录检测实现
大数据-130 - Flink CEP 详解 - CEP开发流程 与 案例实践:恶意登录检测实现
116 0
|
4月前
|
分布式计算 监控 大数据
大数据-129 - Flink CEP 详解 Complex Event Processing - 复杂事件处理
大数据-129 - Flink CEP 详解 Complex Event Processing - 复杂事件处理
92 0
|
5月前
|
SQL 安全 数据处理
揭秘数据脱敏神器:Flink SQL的神秘力量,守护你的数据宝藏!
【9月更文挑战第7天】在大数据时代,数据管理和处理尤为重要,尤其在保障数据安全与隐私方面。本文探讨如何利用Flink SQL实现数据脱敏,为实时数据处理提供有效的隐私保护方案。数据脱敏涉及在处理、存储或传输前对敏感数据进行加密、遮蔽或替换,以遵守数据保护法规(如GDPR)。Flink SQL通过内置函数和表达式支持这一过程。
109 2
|
5月前
|
SQL 大数据 数据处理
奇迹降临!解锁 Flink SQL 简单高效的终极秘籍,开启数据处理的传奇之旅!
【9月更文挑战第7天】在大数据处理领域,Flink SQL 因其强大功能与简洁语法成为开发者首选。本文分享了编写高效 Flink SQL 的实用技巧:理解数据特征及业务需求;灵活运用窗口函数(如 TUMBLE 和 HOP);优化连接操作,优先采用等值连接;合理选择数据类型以减少计算资源消耗。结合实际案例(如实时电商数据分析),并通过定期性能测试与调优,助力开发者在大数据处理中更得心应手,挖掘更多价值信息。
64 1