FlinkSQL之Flink SQL Join二三事

2024-03-24 05:12

本文主要是介绍FlinkSQL之Flink SQL Join二三事,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

​ Flink SQL支持对动态表进行复杂而灵活的连接操作。 为了处理不同的场景,需要多种查询语义,因此有几种不同类型的 Join。默认情况下,joins 的顺序是没有优化的。表的 join 顺序是在 FROM 从句指定的。可以通过把更新频率最低的表放在第一个、频率最高的放在最后这种方式来微调 join 查询的性能。需要确保表的顺序不会产生笛卡尔积,因为不支持这样的操作并且会导致查询失败。

​ Flink Join根据输入源形式不同可以分为双流Join维表Join其他Join多种形式,下面根据大类分别介绍各自特点。

一 双流JOIN

​ 在正式进入FlinkSQL Join场景研究之前,首先我们先介绍一下在FlinkSQL场景下常见的Kafka数据流分类。截止到Flink1.18为止,目前常见的Kafka数据流包括不含键更新的普通Kafka数据流(即Kafka SQL Connector数据流)和包含键更新的Kafka数据流(即Upsert-Kafka SQL Connector数据流)两种。

1 Regular Join

​ Regular join 是最通用的 join 类型。在这种 join 下,join 两侧表的任何新记录或变更都是可见的,并会影响整个 join 的结果。对于流式查询,regular join 的语法是最灵活的,允许任何类型的更新(插入、更新、删除)输入表。 然而,这种操作具有重要的操作意义:Flink 需要将 Join 输入的两边数据永远保持在状态中。 因此,计算查询结果所需的状态可能会无限增长,这取决于所有输入表的输入数据量。你可以提供一个合适的状态 time-to-live (TTL) 配置来防止状态过大。注意:这样做可能会影响查询的正确性。

​ 左右两边流数据都能驱动join,左侧流新加入数据会和右侧流状态中所有匹配记录join上;同理,右侧流新增数据会和左侧流所有匹配记录join上,外连接不会等待,即使Join不上也会即及时输出,待对侧数据到来通过回撤修复数据。

  • Inner Join

    根据 join 限制条件返回一个简单的笛卡尔积。目前只支持 equi-joins,即:至少有一个等值条件。不支持任意的 cross join 和 theta join。

    select t1.order_id    as order_id,t2.product_id  as product_id,t1.create_time as create_time
    from tbl_order t1 
    join tbl_order_product t2 on t1.order_id = t2.order_id 
    ;
    

    Inner join不会产生回撤流,source端可以是Kafka SQL Connector也可以试Upsert-kafka SQL Connector,也可以是混合模式,sink端理论均可以是Kafka Connector,但如果输入端有重复输入,输出端可以设置成Upsert-Kafka SQL Connector接收数据。Upsert-Kafka SQL Connector注意设置主键。

  • outer join

    返回所有符合条件的笛卡尔积(即:所有通过 join 条件连接的行),加上所有外表没有匹配到的行。Flink 支持 LEFT、RIGHT 和 FULL outer joins。目前只支持 equi-joins,即:至少有一个等值条件。不支持任意的 cross join 和 theta join。

    select t1.order_id    as order_id,t2.product_id  as product_id,t1.create_time as create_time
    from tbl_order t1 
    left join tbl_order_product t2 on t1.order_id = t2.order_id 
    ;select t1.order_id    as order_id,t2.product_id  as product_id,t1.create_time as create_time
    from tbl_order t1 
    right join tbl_order_product t2 on t1.order_id = t2.order_id 
    ;select t1.order_id    as order_id,t2.product_id  as product_id,t1.create_time as create_time
    from tbl_order t1 
    full join tbl_order_product t2 on t1.order_id = t2.order_id 
    ;
    

    Outer Join会产生回撤流,source端可以是Kafka SQL Connector也可以是Upsert-kafka SQL Connector,也可以是混合模式,sink端理仅支持设置成Upsert-Kafka SQL Connector接收数据。Upsert-Kafka SQL Connector注意设置主键。

  • Regular Join总结应用模式如下(a代表Append-Only流,u代表Upsert-Kafka流):

    • a join a => a|u

    • u join u => a|u

    • a join u => a|u

    • a left join a => u

    • u left join u => u

    • a left join u => u

2 Interval Join

​ 返回一个符合 join 条件和时间限制的简单笛卡尔积。Interval join 需要至少一个 equi-join 条件和一个 join 两边都包含的时间限定 join 条件。范围判断可以定义成就像一个条件(<, <=, >=, >),也可以是一个 BETWEEN 条件,或者两边表的一个相同类型(即:处理时间 或 事件时间)的时间属性 的等式判断。

​ 下面列举了一些有效的 interval join 时间条件:

  • ltime = rtime
  • ltime >= rtime AND ltime < rtime + INTERVAL '10' MINUTE
  • ltime BETWEEN rtime - INTERVAL '10' SECOND AND rtime + INTERVAL '5' SECOND

​ 对于流式查询,对比 regular join,interval join 只支持有时间属性的Append-Only表。 由于时间属性是递增的,Flink 从状态中移除旧值也不会影响结果的正确性,即interval join会根据间隔自动维护状态大小,不丢弃状态也不会让状态无限增长。

  • Inner join

    select * 
    from tbl_order t1 
    join tbl_shopment t2 on t1.order_id = t2.order_id and t1.create_time between t2.create_time - interval '4' hour and t2.create_time
    ;
    

    ​ 输入源只支持Kafka SQL Connector,不支持任何一方回撤流,这也可以理解,因为Interval Join是有时间属性参与Join的。输出数据可以是Kafka SQL Connector也可以试Upsert-kafka SQL Connector。Upsert-kafka SQL Connector要注意键设计。

  • Outer join

    select * 
    from tbl_order t1 
    left join tbl_shopment t2 on t1.order_id = t2.order_id and t1.create_time between t2.create_time - interval '4' hour and t2.create_time
    ;select * 
    from tbl_order t1 
    right join tbl_shopment t2 on t1.order_id = t2.order_id and t1.create_time between t2.create_time - interval '4' hour and t2.create_time
    ;select * 
    from tbl_order t1 
    full join tbl_shopment t2 on t1.order_id = t2.order_id and t1.create_time between t2.create_time - interval '4' hour and t2.create_time
    ;
    

    ​ 输入端仅至此Kafka SQL Connector,不支持任何一方回撤流,这也可以理解,因为Interval Join是有时间属性参与Outer Join的。输出数据可以是Kafka SQL Connector也可以试Upsert-kafka SQL Connector。Upsert-kafka SQL Connector要注意键设计。

  • 注意点

    • 测试要配置并行度为1,否则右表关联不上数据因为水位线识别不到会而不超时输出;

      executionEnvironment.setParallelism(1);
      
    • left join右表关联不上输出条件

      • 右表关联数据出现触发输出
      • 超时触发器输出关联不上数据
  • Interval Join总结应用模式如下(a代表Append-Only流,u代表Upsert-Kafka流):

    • a join a => a|u

    • a left join a => a|u

3 Temporal Join(Snapshot Join)

​ 时态表(Temporal table)是一个随时间变化的表:在 Flink 中被称为动态表。时态表中的行与一个或多个时间段相关联,所有 Flink 中的表都是时态的(Temporal)。 时态表包含一个或多个版本的表快照,它可以是一个变化的历史表,跟踪变化(例如,数据库变化日志,包含所有快照)或一个变化的维度表,也可以是一个将变更物化的维表(例如,存放最终快照的数据表)。

  • Inner join

    select t1.order_id    as order_id,t1.user_id     as user_id,t2.user_name   as user_name,t1.create_time as create_time
    from tbl_order t1 
    join tbl_user for system_time as of t1.create_time t2 on t1.user_id = t2.user_id 
    ;
    

    特点:

    • 左右两边事件时间属性,标识两侧流join场景,如果处理时间请参考Lookup join;
    • 只支持event-time,如果是processing-time那么就变成join最新版本数据,同Lookup Join;
    • 左表支持append流和upsert流;
    • 右表只支持upsert流;
    • 输出可以是append流或者upsert流;
    • 左表触发计算,右表更新不触发计算;
    • 设置超时时间:tableEnvironment.getConfig().set("table.exec.source.idle-timeout","3s");
  • Left join

    select t1.order_id    as order_id,t1.user_id     as user_id,t2.user_name   as user_name,t1.create_time as create_time
    from tbl_order t1 
    left join tbl_user for system_time as of t1.create_time t2 on t1.user_id = t2.user_id 
    ;
    

    特点:

    • 左右两边事件时间属性,标识两侧流join场景,如果处理时间请参考Lookup join;
    • 只支持event-time,如果是processing-time那么就变成join最新版本数据,同Lookup Join;
    • 左表支持append流和upsert流;
    • 右表只支持upsert流;
    • 输出可以是append流或者upsert流;
    • 左表触发计算,右表更新不触发计算;
    • 设置超时时间:tableEnvironment.getConfig().set("table.exec.source.idle-timeout","3s");
  • Snapshot Join总结应用模式如下(a代表Append-Only流,u代表Upsert-Kafka流):

    • a join u => a|u

    • u join u => u

    • a left join u => a|u

    • u left join u => u

4 Window Join

​ 窗口关联就是增加时间维度到关联条件中。在此过程中,窗口关联将两个流中在同一窗口且符合 join 条件的元素 join 起来。窗口关联的语义和DataStream window join相同。

​ 在流式查询中,与其他连续表上的关联不同,窗口关联不产生中间结果,只在窗口结束产生一个最终的结果。另外,窗口关联会清除不需要的中间状态。

​ 通常,窗口关联和窗口表值函数一起使用。而且,窗口关联可以在其他基于窗口表值函数的操作后使用,例如窗口聚合,窗口 Top-N和窗口关联。

​ 目前,窗口关联需要在 join on 条件中包含两个输入表的 window_start 等值条件和 window_end 等值条件。

​ 窗口关联支持 INNER/LEFT/RIGHT/FULL OUTER/ANTI/SEMI JOIN。

  • 语法

    select ...
    from l [left|right|full outer] join r -- l and r are relations applied windowing TVF
    on l.window_start = r.window_start and l.window_end = r.window_end and ...
    
  • 注意

    • 当前版本窗口Join必须同时指定window_start和window_end等值条件

    • 窗口Join不支持源是upsert流的情况

  • 限制

    • Join 子句的限制

    ​ 目前,窗口关联需要在 join on 条件中包含两个输入表的 window_start 等值条件和 window_end 等值条件。未来,如果是滚动或滑动窗口,只需要在 join on 条件中包含窗口开始相等即可。

    • 输入的窗口表值函数的限制

    ​ 目前,关联的左右两边必须使用相同的窗口表值函数。这个规则在未来可以扩展,比如:滚动和滑动窗口在窗口大小相同的情况下 join。

    • 窗口表值函数之后直接使用窗口关联的限制

    ​ 目前窗口关联支持作用在滚动(TUMBLE)、滑动(HOP)和累积(CUMULATE)窗口表值函数之上,但是还不支持会话窗口(SESSION)。

  • Snapshot Join总结应用模式如下(a代表Append-Only流,u代表Upsert-Kafka流):

    • a join a => a|u

    • a left join a => a|u

二 维表JOIN

5 Lookup Join(processing-time temporal join)

​ lookup join 通常用于使用从外部系统查询的数据来丰富表。join 要求一个表具有处理时间属性,另一个表由查找源连接器(lookup source connnector)支持。通常使用基于处理时间的流表与外部版本表(例如 mysql、hbase)的最新版本相关联(即processing-time temporal join 常常用在使用外部系统来丰富流的数据)。

​ 通过定义一个处理时间属性,这个 join 总是返回最新的值。可以将 build side 中被查找的表想象成一个存储所有记录简单的 HashMap<K,V>。 这种 join 的强大之处在于,当无法在 Flink 中将表具体化为动态表时,它允许 Flink 直接针对外部系统工作。

​ Join操作由流端触发,当新增一个流数据,会查询外部DB映射,获取数据补全后发出结果数据。

  • inner join

    select t1.order_id    as order_id,t1.user_id     as user_id,t2.user_name   as user_name,t1.create_time as create_time
    from tbl_order t1 
    join tbl_user for system_time as of t1.create_time t2 on t1.user_id = t2.user_id 
    ;
    

    特点:

    • Lookup join只支持inner join和left join;
    • 源必须声明处理时间,即row_time as proctime(),如果源声明为事件时间,那么要走Snapshot join方式;
    • 源支持kafka和upsert-kafka连接器
    • 输出支持kafka和upsert-kafka连接器
    • 查询外部表注意使用异步IO/Cache特性优化外表查询性能
  • Left join

    select t1.order_id    as order_id,t1.user_id     as user_id,t2.user_name   as user_name,t1.create_time as create_time
    from tbl_order t1 
    left join tbl_user for system_time as of t1.create_time t2 on t1.user_id = t2.user_id 
    ;
    

    特点:

    • Lookup join只支持inner join和left join;
    • 源必须声明处理时间,即row_time as proctime(),如果源声明为事件时间,那么要走Snapshot join方式;
    • 源支持kafka和upsert-kafka连接器
    • 输出支持kafka和upsert-kafka连接器
    • 查询外部表注意使用异步IO/Cache特性优化外表查询性能
  • Lookup Join总结应用模式如下(a代表Append-Only流,s代表外表静态表):

    • a join s => a|u

    • u join s => a|u

    • a left join s => a|u

    • u left join s => a|u

三 其他JOIN

6 Array Expansion

​ 对于输入的包含数组列的单行数据,返回给定数组中每个元素的新行,拆分后的数据除解析数组元素外,其他元素与原始行数据一致。

selectorder_id,order_tag,tag
from tbl_order_source cross join unnest(order_tag) as t(tag)
;

特征:

  • 输入数据可以是Append或者Upsert
  • 输出数据可以是Append或者Upsert

7 Table Function

​ 将表与表函数的结果联接。左侧(外部)表的每一行都与表函数的相应调用产生的所有行相连接。用户自定义表函数必须在使用前注册。

​ 对于是inner join,如果表函数调用返回一个空结果,那么左表的这行数据将不会输出。对于left join,如果表函数调用返回了一个空结果,则保留相应的行,并用空值填充未关联到的结果。当前,针对 lateral table 的 left outer join 需要 ON 子句中有一个固定的 TRUE 连接条件。

select order_id,order_tag,tag
from tbl_order_source
left join lateral table(table_func(order_tag)) t(tag) on true
;

特征:

  • 输入数据可以是Append或者Upsert
  • 输出数据可以是Append或者Upsert

这篇关于FlinkSQL之Flink SQL Join二三事的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!


原文地址:
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.chinasem.cn/article/840545

相关文章

MySQL 8 中的一个强大功能 JSON_TABLE示例详解

《MySQL8中的一个强大功能JSON_TABLE示例详解》JSON_TABLE是MySQL8中引入的一个强大功能,它允许用户将JSON数据转换为关系表格式,从而可以更方便地在SQL查询中处理J... 目录基本语法示例示例查询解释应用场景不适用场景1. ‌jsON 数据结构过于复杂或动态变化‌2. ‌性能要

MySQL字符串常用函数详解

《MySQL字符串常用函数详解》本文给大家介绍MySQL字符串常用函数,本文结合实例代码给大家介绍的非常详细,对大家学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录mysql字符串常用函数一、获取二、大小写转换三、拼接四、截取五、比较、反转、替换六、去空白、填充MySQL字符串常用函数一、

MySQL中比较运算符的具体使用

《MySQL中比较运算符的具体使用》本文介绍了SQL中常用的符号类型和非符号类型运算符,符号类型运算符包括等于(=)、安全等于(=)、不等于(/!=)、大小比较(,=,,=)等,感兴趣的可以了解一下... 目录符号类型运算符1. 等于运算符=2. 安全等于运算符<=>3. 不等于运算符<>或!=4. 小于运

虚拟机Centos7安装MySQL数据库实践

《虚拟机Centos7安装MySQL数据库实践》用户分享在虚拟机安装MySQL的全过程及常见问题解决方案,包括处理GPG密钥、修改密码策略、配置远程访问权限及防火墙设置,最终通过关闭防火墙和停止Net... 目录安装mysql数据库下载wget命令下载MySQL安装包安装MySQL安装MySQL服务安装完成

MySQL进行数据库审计的详细步骤和示例代码

《MySQL进行数据库审计的详细步骤和示例代码》数据库审计通过触发器、内置功能及第三方工具记录和监控数据库活动,确保安全、完整与合规,Java代码实现自动化日志记录,整合分析系统提升监控效率,本文给大... 目录一、数据库审计的基本概念二、使用触发器进行数据库审计1. 创建审计表2. 创建触发器三、Java

MySQL逻辑删除与唯一索引冲突解决方案

《MySQL逻辑删除与唯一索引冲突解决方案》本文探讨MySQL逻辑删除与唯一索引冲突问题,提出四种解决方案:复合索引+时间戳、修改唯一字段、历史表、业务层校验,推荐方案1和方案3,适用于不同场景,感兴... 目录问题背景问题复现解决方案解决方案1.复合唯一索引 + 时间戳删除字段解决方案2:删除后修改唯一字

Zabbix在MySQL性能监控方面的运用及最佳实践记录

《Zabbix在MySQL性能监控方面的运用及最佳实践记录》Zabbix通过自定义脚本和内置模板监控MySQL核心指标(连接、查询、资源、复制),支持自动发现多实例及告警通知,结合可视化仪表盘,可有效... 目录一、核心监控指标及配置1. 关键监控指标示例2. 配置方法二、自动发现与多实例管理1. 实践步骤

MySQL 主从复制部署及验证(示例详解)

《MySQL主从复制部署及验证(示例详解)》本文介绍MySQL主从复制部署步骤及学校管理数据库创建脚本,包含表结构设计、示例数据插入和查询语句,用于验证主从同步功能,感兴趣的朋友一起看看吧... 目录mysql 主从复制部署指南部署步骤1.环境准备2. 主服务器配置3. 创建复制用户4. 获取主服务器状态5

SpringBoot中六种批量更新Mysql的方式效率对比分析

《SpringBoot中六种批量更新Mysql的方式效率对比分析》文章比较了MySQL大数据量批量更新的多种方法,指出REPLACEINTO和ONDUPLICATEKEY效率最高但存在数据风险,MyB... 目录效率比较测试结构数据库初始化测试数据批量修改方案第一种 for第二种 case when第三种

MySql基本查询之表的增删查改+聚合函数案例详解

《MySql基本查询之表的增删查改+聚合函数案例详解》本文详解SQL的CURD操作INSERT用于数据插入(单行/多行及冲突处理),SELECT实现数据检索(列选择、条件过滤、排序分页),UPDATE... 目录一、Create1.1 单行数据 + 全列插入1.2 多行数据 + 指定列插入1.3 插入否则更