vps网站如何绑定多个域名,广阳网站制作,建设营销型网站有哪些步骤,北京营销策划公司有哪些 Flink SQL支持对动态表进行复杂而灵活的连接操作。 为了处理不同的场景#xff0c;需要多种查询语义#xff0c;因此有几种不同类型的 Join。默认情况下#xff0c;joins 的顺序是没有优化的。表的 join 顺序是在 FROM 从句指定的。可以通过把更新频率最低的表放在第一个… 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 rtimeltime rtime AND ltime rtime INTERVAL 10 MINUTEltime BETWEEN rtime - INTERVAL 10 SECOND AND rtime INTERVAL 5 SECOND 对于流式查询对比 regular joininterval 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 中被查找的表想象成一个存储所有记录简单的 HashMapK,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
