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

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Leetcode9_回文数

Leetcode9_回文数

1.leetcode原题链接&#xff1a;. - 力扣&#xff08;LeetCode&#xff09; 2.题目描述 给你一个整数 x &#xff0c;如果 x 是一个回文整数&#xff0c;返回 true &#xff1b;否则&#xff0c;返回 false 。 回文数 是指正序&#xff08;从左向右&#xff09;和倒序&…
阅读更多...
llama.cpp 编译安装@Ubuntu

llama.cpp 编译安装@Ubuntu

在Kylin 和Ubuntu编译llama.cpp &#xff0c;具体参考&#xff1a;llama模型c语言推理FreeBSD-CSDN博客 现在代码并编译&#xff1a; git clone https://github.com/ggerganov/llama.cpp cd llama.cpp mkdir build cd build cmake .. cmake --build . --config Release# 可选…
阅读更多...
Spring 基于XML的DI

Spring 基于XML的DI

例如以下类&#xff1a; teanDao是没有new的&#xff0c;所以直接用的时候会报空指针异常&#xff0c;所以再Spring自动创建该Service的时候要给teamDao注入&#xff0c;也就是赋值吧。 因为TeamDao这个对象是要用的&#xff0c;所以这个对象也是要创建的&#xff0c;但是要在…
阅读更多...
【 MyBatis 】| 关于多表联查返回 List 集合只查到一条的 BUG

【 MyBatis 】| 关于多表联查返回 List 集合只查到一条的 BUG

目录 一. &#x1f981; 写在前面二. &#x1f981; 探索过程2.1 开端 —— 开始写 bug2.2 发展 —— bug 完成2.3 高潮 —— bug探究2.4 结局 —— 效果展示 三. &#x1f981; 写在最后 一. &#x1f981; 写在前面 今天又是 BUG 气满满的一天&#xff0c;一个 xxxMapper.xm…
阅读更多...
spark核心概念

spark核心概念

DAG 所谓DAG就是有向无环图&#xff0c;其实就是个无环的流程&#xff0c;Spark的核心是根据RDD来实现的&#xff0c;Spark Scheduler!则为Spark核心实现的重要一环&#xff0c;其作用就是任务调度。Spark的任务调度就是如何组织任务去处理RDD中每个分区的数据&#xff0c;根据…
阅读更多...
Day53：WEB攻防-XSS跨站SVGPDFFlashMXSSUXSS配合上传文件添加脚本

Day53：WEB攻防-XSS跨站SVGPDFFlashMXSSUXSS配合上传文件添加脚本

目录 MXSS UXSS&#xff1a;Universal Cross-Site Scripting HTML&SVG&PDF&SWF-XSS&上传&反编译(有几率碰到) SVG-XSS PDF-XSS Python生成XSS Flash-XSS 知识点&#xff1a; 1、XSS跨站-MXSS&UXSS 2、XSS跨站-SVG制作&配合上传 3、XSS跨站-…
阅读更多...
@分布之间的关系 --------从分布之间的关系来理解随机现象、进而理解概率论

@分布之间的关系 --------从分布之间的关系来理解随机现象、进而理解概率论

分布之间的关系 --------从分布之间的关系来理解随机现象、进而理解概率论 文章目录 关系分类 关系分类 概率分布之间的关系分两类: 变换、变形 作用于随机变量、概率分布&#xff0c;产生新分布 例如 随机变量的和、随机变量的乘积、随机变量的函数极限分布 当某些参数取极…
阅读更多...
Docker进阶：使用Docker部署Harbor私有镜像仓库

Docker进阶：使用Docker部署Harbor私有镜像仓库

Docker进阶&#xff1a;使用Docker部署Harbor私有镜像仓库 1、安装Docker和Docker Compose1、安装Docker、Docker Compose2、验证Docker和Docker Compose是否成功安装3、先启动运行docker服务 2、下载并配置Harbor1、下载最新版本的Harbor离线安装包2、配置Harbor的主机名和管理…
阅读更多...
一步步指导：在Chrome中安装Vue插件vue-devtools

一步步指导：在Chrome中安装Vue插件vue-devtools

一步步指导&#xff1a;在Chrome中安装Vue插件vue-devtools 1. 引言1.1. 关于Vue.js1.2. 为何使用vue-devtools 2. vue-devtools简介2.1. vue-devtools的功能2.2. 兼容性和需求 3. 如何在Chrome中安装vue-devtools3.1. 访问Chrome网上应用店3.2. 搜索并找到vue-devtools3.3. 安…
阅读更多...
Linux共享网络给其它主机

Linux共享网络给其它主机

Linux共享网络给其它主机 文章目录 前言设置有网主机转发设置无网主机接入ip route设置route设置固化配置 总结 前言 本文主要讲解如何快速配置共享网络给同网段内其它主机&#xff0c;主要有如下几种情况&#xff1a; 本机有无线和有线网卡&#xff0c;无线能上网有线不能上…
阅读更多...
京东云搭建幻兽帕鲁Palworld多人游戏联机服务器教程，1分钟开服

京东云搭建幻兽帕鲁Palworld多人游戏联机服务器教程，1分钟开服

使用京东云服务器搭建幻兽帕鲁Palworld游戏联机服务器教程&#xff0c;非常简单&#xff0c;京东云推出幻兽帕鲁镜像系统&#xff0c;镜像直接选择幻兽帕鲁镜像即可一键自动部署&#xff0c;不需要手动操作&#xff0c;真正的新手0基础部署幻兽帕鲁&#xff0c;阿腾云atengyun.…
阅读更多...
uni-app（使用阿里图标）

uni-app（使用阿里图标）

1.注册阿里矢量图标库 注册阿里图标库账号并登录&#xff0c;https://www.iconfont.cn/ 2.加入购物车 搜索适合自己的图标&#xff0c;加入购物车&#xff0c;如下图&#xff1a; 3.加入项目 我的->资源管理->我的项目->创建项目&#xff0c;然后返回购物车&#…
阅读更多...
Kafka集群安装与配置

Kafka集群安装与配置

Kafka 提醒&#xff1a;需要先安装zookeeper后才可以安装kafka 安装 官方下载地址&#xff1a;http://kafka.apache.org/downloads.html 解压安装包 tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/修改解压后的文件名称 mv kafka_2.12-3.0.0/ kafka进入到/opt/module/k…
阅读更多...
深度学习论文阅读之【Distilling the Knowledge in a Neural Network】提炼神经网络中的知识

深度学习论文阅读之【Distilling the Knowledge in a Neural Network】提炼神经网络中的知识

论文&#xff1a;link 代码&#xff1a;link 摘要 提高几乎所有机器学习算法性能的一个非常简单的方法是在相同的数据上训练许多不同的模型&#xff0c;然后对它们的预测进行平均[3]。不幸的是&#xff0c;使用整个模型集合进行预测非常麻烦&#xff0c;并且计算成本可能太高&…
阅读更多...
HTML文本信息

HTML文本信息

标题 使用h1~h6标签定义标题。通常一个HTML网页只有一个主标题和副标题&#xff0c;主标题和副标题分别使用h1和h2表示。 <h1>主标题</h1> <h2>副标题</h2><p>正文</p>段落 p元素用来表示段落文本。通常用来显示大片的文字。每一个p元素…
阅读更多...
中国信通院 X StarRocks金融用户社区正式成立

中国信通院 X StarRocks金融用户社区正式成立

在国家战略的推动下&#xff0c;开源技术正逐渐成为金融行业创新发展的重要驱动力。2024 年 3 月 26 日&#xff0c;中国信息通信研究院 X StarRocks 金融用户社区&#xff08;以下简称“社区”&#xff09;正式成立&#xff0c;这一举措旨在深化国内金融领域的开源生态建设&am…
阅读更多...
粗略总结AI大模型学习需要了解的要点

粗略总结AI大模型学习需要了解的要点

目录 一、概念简介 二、兴起原因 三、相关要点 四、不足之处 五、总结 一、概念简介 AI大模型学习是指利用大规模数据集和强大计算能力进行深度学习模型的训练。随着数据的爆炸式增长和计算资源的提升&#xff0c;AI大模型学习成为了现代人工智能研究的重要方向。 二、兴起…
阅读更多...
单元测试11213123231313131231231231

单元测试11213123231313131231231231

使用技术 junit Mockito s[romg 示例代码&#xff1a; SpringBootTest(classes启动类.class) public class AbstractTes{ MockBean protected A a; } AutoConfigureMockMvc(printOnlyOnFailure false) public abstract class AbstractWebTes extends AbstractTes imple…
阅读更多...
使用pytorch构建一个初级的无监督的GAN网络模型

使用pytorch构建一个初级的无监督的GAN网络模型

在这个系列中将系统的构建GAN及其相关的一些变种模型&#xff0c;来了解GAN的基本原理。本片为此系列的第一篇&#xff0c;实现起来很简单&#xff0c;所以不要期待有很好的效果出来。 第一篇我们搭建一个无监督的可以生成数字 (0-9) 手写图像的 GAN&#xff0c;使用MINIST数据…
阅读更多...
精准测试——BCEL字节码检测

精准测试——BCEL字节码检测

精准测试是通过源代码变更分析&#xff0c;确定改动代码影响的范围&#xff0c;从而进行针对性测试&#xff0c;进一步提升测试效率。不仅如此&#xff0c;精准测试还可以将测试用例与程序代码之间的逻辑映射关系建立起来&#xff0c;采集测试过程执行的代码逻辑及测试数据。怎…
阅读更多...
最新文章

Copyright @ 2022~2023