Flink Sql:四种Join方式详解(基于flink1.15官方文档)

JOINs

flink sql主要有四种连接方式,分别是Regular Joins、Interval Joins、Temporal Joins、lookup join

1、Regular Joins(常规连接 )

这种连接方式和hive sql中的join是一样的,包括inner join,left join,right join,full join

1、指定数据源建立students表
CREATE TABLE students (id STRING,name STRING,age INT,sex STRING,clazz STRING
) WITH ('connector' = 'kafka','topic' = 'students', -- 指定topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- 指定kafka集群列表'properties.group.id' = 'testGroup', -- 指定消费者组'scan.startup.mode' = 'latest-offset', -- 指定读取数据的位置为最新生成的数据'format' = 'csv' -- 指定数据的格式
);2、kafka生产students表数据
kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic students
1500100001,施笑槐,22,女,文科六班
1500100002,吕金鹏,24,男,文科六班
1500100003,单乐蕊,22,女,理科六班3、创建关联表scores
CREATE TABLE scores (sid STRING,   cid STRING,     --学科idscore INT     
) WITH ('connector' = 'kafka','topic' = 'scores', -- 指定topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- 指定kafka集群列表'properties.group.id' = 'testGroup', -- 指定消费者组'scan.startup.mode' = 'latest-offset', -- 指定读取数据的位置'format' = 'csv' -- 指定数据的格式
);4、kafka生产scores数据
kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic scores
1500100001,1000001,98
1500100001,1000002,56
1500100002,1000001,139
1500100002,1000002,102
1500100004,1000001,42
1500100004,1000002,142-- inner jion   两边数据都不为null的才会关联
select 
a.id,a.name,b.sid,b.score
from 
students as a
inner join
scores as b
on a.id=b.sid;-- left join/right join    保证左边/右边数据的完整性
select 
a.id,a.name,b.sid,b.score
from 
students as a
right join
scores as b
on a.id=b.sid;-- full join       保证两边数据的完整性
select 
a.id,a.name,b.sid,b.score
from 
students as a
full join
scores as b
on a.id=b.sid;-- 注意:
-- 常规连接,会将两个表的数据一直保存在状态中,时间长了,状态会越来越大,导致任务执行失败,通常在批处理中使用,因为批处理没有状态这个概念。为了避免状态过大可能会导致的任务失败问题,我们可以设置状态有效期
-- 状态有效期,状态在flink中保存的时间,但是如果sql中除了关联操作还有聚合这样也需要将数据保存在状态中的操作,状态有效期设置的太短可能会让聚合这样的操作失败,设置的太长延迟也会增加。所以,状态保留多久需要根据实际业务分析
SET 'table.exec.state.ttl' = '20000';
设置该参数后,那么只有在20秒内到达的数据才会被保存到状态中进行关联。

inner join结果:

left join 结果:

right join结果:

full join结果:

2、Interval Joins(间隔连接

Interval Joins:在一段时间内关联

对于流式查询,与常规连接相比,间隔连接仅支持具有时间属性的追加表。由于时间属性是拟单调递增的,因此 Flink 可以从其状态中删除旧值,而不会影响结果的正确性。

这种方式可以变相弥补Regular Joins中时间长了状态过大的问题。

CREATE TABLE students_proctime (id STRING,name STRING,age INT,sex STRING,clazz STRING,proctime AS PROCTIME()
) WITH ('connector' = 'kafka','topic' = 'students', -- 指定topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- 指定kafka集群列表'properties.group.id' = 'testGroup', -- 指定消费者组'scan.startup.mode' = 'latest-offset', -- 指定读取数据的位置'format' = 'csv' -- 指定数据的格式
);kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic students
1500100001,施笑槐,22,女,文科六班
1500100002,吕金鹏,24,男,文科六班
1500100003,单乐蕊,22,女,理科六班CREATE TABLE scores_proctime (sid STRING,cid STRING,score INT,proctime AS PROCTIME()
) WITH ('connector' = 'kafka','topic' = 'scores', -- 指定topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- 指定kafka集群列表'properties.group.id' = 'testGroup', -- 指定消费者组'scan.startup.mode' = 'latest-offset', -- 指定读取数据的位置'format' = 'csv' -- 指定数据的格式
);
kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic scores
1500100001,1000001,98
1500100001,1000002,56
1500100002,1000001,139
1500100002,1000002,102
1500100004,1000001,42
1500100004,1000002,142select a.id,a.name,b.sid,b.score from 
students_proctime a, scores_proctime b
where a.id=b.sid
-- a表的时间需要在b表时间10秒内或b表的时间需要在a表时间10秒内
and (a.proctime BETWEEN b.proctime - INTERVAL '10' SECOND AND b.proctimeor b.proctime BETWEEN a.proctime - INTERVAL '10' SECOND AND a.proctime
);

3、Temporal Joins(时态连接)

这种关联方式是专门用来关联时态表的。

  • Temporal Joins(时态连接)是在流式计算或数据处理中,对两个或多个随时间变化的表(也称为动态表或时态表)进行连接的操作。这些表包含随时间变化的数据,并且行与一个或多个时态周期相关联。

在我们生活中最常见的时态表就是汇率表,汇率随着时间变化而变化。

 

案例:

例如,假设我们有一张订单表,每张订单的价格都采用不同的货币。为了正确地将此表标准化为单一货币(如美元),每张订单都需要与下订单时相应的货币兑换率相结合。

1、创建订单表
CREATE TABLE orders (order_id    STRING,price       DECIMAL(32,2),currency    STRING,    --币种order_time  TIMESTAMP(3),WATERMARK FOR order_time AS order_time
) WITH ('connector' = 'kafka','topic' = 'orders', -- 指定topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- 指定kafka集群列表'properties.group.id' = 'testGroup', -- 指定消费者组'scan.startup.mode' = 'latest-offset', -- 指定读取数据的位置'format' = 'csv' -- 指定数据的格式
);2、订单表数据
kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic orders
o_001,1,EUR,2024-06-06 12:00:00
o_002,100,EUR,2024-06-06 12:00:07
o_003,200,EUR,2024-06-06 12:00:16
o_004,10,EUR,2024-06-06 12:00:21
o_005,20,EUR,2024-06-06 12:00:253、创建汇率表
CREATE TABLE currency_rates (currency STRING,conversion_rate DECIMAL(32, 2),update_time TIMESTAMP(3),WATERMARK FOR update_time AS update_time,PRIMARY KEY(currency) NOT ENFORCED -- 主键,区分不同的汇率
) WITH ('connector' = 'kafka','topic' = 'currency_rates1', -- 指定topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- 指定kafka集群列表'properties.group.id' = 'testGroup', -- 指定消费者组'scan.startup.mode' = 'earliest-offset', -- 指定读取数据的位置'format' = 'canal-json' -- 指定数据的格式
);4、向汇率表中添加数据
insert into currency_rates
values
('EUR',0.12,TIMESTAMP'2024-06-06 12:00:00'),
('EUR',0.11,TIMESTAMP'2024-06-06 12:00:09'),
('EUR',0.15,TIMESTAMP'2024-06-06 12:00:17'),
('EUR',0.14,TIMESTAMP'2024-06-06 12:00:23');kafka-console-consumer.sh --bootstrap-server  master:9092,node1:9092,node2:9092 --from-beginning --topic currency_rates-- 使用常规关联方式关联时态表只能关联到最新的数据
select 
a.price,a.order_time,b.conversion_rate,b.update_time
from 
orders as a
join
currency_rates as b
on a.currency=b.currency;-- 时态表join
-- FOR SYSTEM_TIME AS OF a.order_time: 使用a表的时间到b表中查询对应时间段的数据
select 
a.price,a.order_time,b.conversion_rate,b.update_time
from 
orders as a
join
currency_rates FOR SYSTEM_TIME AS OF a.order_time as b 
on a.currency=b.currency;

常规join结果:

时态join结果:

4、lookup join(查找连接

Lookup Join,也称为维表 Join,通常用于从外部系统查询的数据表。连接要求一个表具有处理时间属性,另一个表由查找源连接器支持。

具体来说:

lookup join用于流表(动态表)关联维度表

流表:动态表

维度表:不怎么变化的变,维度表的数据一般可以放在hdfs或者mysql等外部数据源


扩展:流表、事实表、维度表

-- 流表(动态表)
1、流表的数据来源通常是实时数据流,这些数据流可以来自各种数据源,如 Kafka、RabbitMQ、Kinesis 等。Flink可以通过数据源连接器(Source Connectors)将这些实时数据流接入到 Flink 系统中
2、与传统数据库中的表不同,流表的行是动态生成的,随着数据流的持续产生而不断增加-- 维度表
1、主要提供数据的分析角度,包含了描述业务环境的属性信息,如时间、地理、产品等。
2、维度表:通常比较宽(包含多个属性列),但行数相对较少,因为维度表中的每一行通常代表一个具体的业务实体或类别,如一个商品、一个客户、一个日期等。
3、维度表与事实表之间通过外键相关联,共同构成了星型模型或雪花模型。事实表中的外键用于与维度表中的主键相匹配,从而提供数据的上下文和分类信息。
4、维度表存储的是对数据的描述性信息,这些信息通常不随时间变化,或者变化不频繁。例如,商品的品牌、型号、颜色等属性一旦确定后很少会发生变化。但在某些情况下,如新产品上市或促销活动,可能需要更新维度表以添加新的维度成员。-- 事实表
1、存储了实际的数据度量值,如销售额、订单数量等。事实表是数据分析的核心,包含了所有用于分析的数据指标。
2、通常比较窄(包含较少的列),但行数非常多,因为事实表中的每一行通常代表一个具体的事件或交易,如一个订单、一次点击等。
3、事实表存储的是度量数据(即指标),这些数据会随时间变化,并且经常需要被汇总和分析。例如,销售额、订单数量、点击量等指标会随着业务活动的进行而不断更新。
4、事实表的数据更新频率通常较高,因为事实数据会随着业务活动的进行而不断产生。例如,每当有新的订单产生时,都需要在事实表中插入一条新的记录。

 

1、创建分数表
CREATE TABLE scores (sid INT,cid STRING,score INT,proctime AS PROCTIME()
) WITH ('connector' = 'kafka','topic' = 'scores', -- 指定topic'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092', -- 指定kafka集群列表'properties.group.id' = 'testGroup', -- 指定消费者组'scan.startup.mode' = 'latest-offset', -- 指定读取数据的位置'format' = 'csv' -- 指定数据的格式
);2、生产分数表数据
kafka-console-producer.sh --broker-list master:9092,node1:9092,node2:9092 --topic scores
1500100001,1000001,98
1500100001,1000003,1373、建立学生表,我们将学生表当作维度表放在mysql中
CREATE TABLE students_test (id INT,name STRING,age INT,gender STRING,clazz STRING
) WITH ('connector' = 'jdbc','url' = 'jdbc:mysql://master:3306/bigdata29','table-name' = 'students_test','username' ='root','password' = '123456','lookup.cache.max-rows' = '1000', -- 最大缓存行数'lookup.cache.ttl' ='10000' -- 缓存过期时间
);学生表数据
1500100001,施笑槐,22,女,文科六班-- 使用常规关联方式
-- 维表的数据只在任务启动的时候读取一次,后面不再实时读取,
-- 只能关联到任务启动时读取的数据
-- 一旦mysql中的学生表更新数据,但是关联的学生表数据却是任务启动时从mysql读取的,这就有错误了,lookup join可以解决该问题。
select a.sid,a.score,b.id,b.name from
scores as a
left join
students_test  as b
on a.sid=b.id;-- lookup join
-- 当流表每来一条数据时,使用关联字段到维表的数据源中查询
-- 优点:实时更新数据源,准确性高
-- 缺点:每一次都需要查询数据库,性能会降低
select a.sid,a.score,b.id,b.name from
scores as a
left join
students_test FOR SYSTEM_TIME AS OF a.proctime as b
on a.sid=b.id;

此时我们修改更新mysql中的学生表数据

修改之前

修改后:

常规关联结果:

look up关联结果:

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

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

相关文章

visual studio下载安装

1、下载网址:下载 Visual Studio Tools - 免费安装 Windows、Mac、Linux 选择下载“社区” 2、下载好之后,安装在非系统盘上,在下面这个界面上,大家可以把自己需要的都勾选上,然后更改安装地址 安装完即可

python中列表结构在点云数据处理中用法

1、前言 Python中的列表(list)是一种可变的序列类型,用于存储集合数据。列表用途非常广泛,包括但不限于以下几个方面: 存储集合数据:列表用于存储一系列有序的元素,这些元素可以是任何数据类型&…

开源-Docker部署Cook菜谱工具

开源-Docker部署Cook菜谱工具 文章目录 开源-Docker部署Cook菜谱工具介绍资源列表基础环境一、安装Docker二、配置加速器三、查看Docker版本四、拉取cook镜像五、部署cook菜谱工具5.1、创建cook容器5.2、查看容器运行状态5.3、查看cook容器日志 六、访问cook菜谱服务6.1、访问c…

PRP和SGL 你了解吗?

一直想总结一下,PRP和SGL,网上也有不少资料,nvme官方spec只用了六七页解释了这俩货,还把寄存器如何操作也说明白了,作总结的目的是让自己对其更加深入了解 首先,SSD是用来保存数据,不是读就是写…

JavaWeb6 Tomcat+postman请求、响应

Web服务器 对HTTP协议操作进行封装,简化web程序开发 部署web项目,对外提供网上信息浏览服务 Tomcat 轻量级web服务器,支持servlet,jsp等少量javaEE规范 也被称为web容器,servlet容器 Springboot有内置Tomcat nginx…

制造业泄密如何防范?应用迅软DSE加密软件能解决哪些问题?

项目背景 某公司电子技术产品广泛应用于航天、航空、航海、遥测、导航、雷达、电子对抗、通信等高端领域。内部会有各种各样的研发核心数据流转在不同的岗位之间,这些核心数据一旦出现信息泄密或篡改数据的情况,将会给企业带来不可估量的经济损失&#…

cloudflare worker访问自己的网站显示521问题解决

写在前面:如果你的网站不是在80端口上运行的,开一下80端口可能就行了… 1.在cloudlare上添加域名 前文搭建了自己的DNS服务器(DNS服务器搭建),现在想通过自己的DNS服务器解析域名,需要四步: 添…

基于FreeRTOS+STM32CubeMX+LCD1602+MCP6S21(SPI接口)的单通道模拟可编程增益放大器Proteus仿真

一、简介: MCP6S21是单通道模拟可 编程增益放大器(Programmable Gain Amplifiers, PGA)。它们可配置为输出 +1 V/V 到 +32 V/V 之间的增 益。串行接口也可以将 PGA 置为关断模式,以降低 功耗。这些 PGA 针对高速度、低失调电压和单电源操 作进行了优化,具有轨到轨输入和输…

使用ZIP包安装MySQL及配置教程

在本教程中,我们将指导您完成使用ZIP包安装MySQL的过程,并对配置文件进行必要的修改,以及解决可能遇到的问题。本示例以MySQL 5.7.44为例,但步骤同样适用于其他版本如MySQL 8.3.0等。请根据实际需要选择适合的版本下载&#xff1a…

【Nginx系列】分发算法

文章目录 一、分发算法介绍二、nginx集群默认算法三、nginx业务服务器状态四、nginx集群默认算法测试实验环境实验拓扑4.1、轮询算法4.2、基于权重4.3、基于ip_hash分发4.4、基于url的hash 🌈你好呀!我是 山顶风景独好 🎈欢迎踏入我的博客世界…

【启明智显方案分享】ESP32-S3与GPT AI融合的智能问答嵌入式设备应用解决方案

一、引言 随着物联网(IoT)和人工智能(AI)技术的飞速发展,嵌入式设备正逐渐变得智能化。本解决方案是启明智显通过结合ESP32-S3的低功耗、高性能特性和GPT(Generative Pre-trained Transformer)…

快手AI算法岗,50W年包羡慕到流泪

今天在脉脉上看到一个应届毕业生offer选择的帖子,简直羡慕到流泪。 刚毕业就拿到了两大公司的AI算法岗的offer,而且薪资待遇都非常不错,只能说:优秀的人到哪里都是榜样。 先看下这两个offer。 第一个是中信银行的AI算法。 年包…

改进YOLOv8 | 主干网络篇 | YOLOv8 更换主干网络之 StarNet | 《重写星辰⭐》

本改进已集成到 YOLOv8-Magic 框架。 论文地址:https://arxiv.org/abs/2403.19967 论文代码:https://github.com/ma-xu/Rewrite-the-Stars 最近的研究引起了人们对“星形运算”(按元素乘法)在网络设计中未被充分利用的潜力的关注。虽然直观的解释很多,但其应用的基本原理…

网络的下一次迭代:AVS 将为 Web2 带去 Web3 的信任机制

撰文:Sumanth Neppalli,Polygon Ventures 编译:Yangz,Techub News 本文来源香港Web3媒体:Techub News AVS (主动验证服务)将 Web2 的规模与 Web3 的信任机制相融合,开启了网络的下…

OpenCV 的模板匹配

OpenCV中的模板匹配 模板匹配(Template Matching)是计算机视觉中的一种技术,用于在大图像中找到与小图像(模板)相匹配的部分。OpenCV提供了多种模板匹配的方法,主要包括基于相关性和基于平方差的匹配方法。…

前端数据模拟Mock.js

新建mock-demo的项目,安装npm install mockjs 新建index.js //引入mockjs import Mock from mockjs //设置延迟时间 // Mock.setup({ // timeout:4000 // }) //使用mockjs模拟数据 Mock.mock(/product/search,{"ret":0,"data":{"mtim…

动手学深度学习33 单机多卡并行

单机多卡并行 更多的芯片 https://courses.d2l.ai/zh-v2/assets/pdfs/part-2_2.pdf 多GPU训练 https://courses.d2l.ai/zh-v2/assets/pdfs/part-2_3.pdf 当transformer模型很大,有100GB的时候只能用模型并行。 数据并行,拿的参数是完整的&#xff1f…

JavaWeb5 SpringBoot+HTTP协议

Spring Spring Boot 非常快速构建应用程序,简化开发 (1)创建Springboot工程,勾选web开发依赖 创建好的目录,并将没用多余的删掉了 (2)定义请求处理类,并添加方法 创建请求处理类…

机器学习分类及算法

1. 深度学习 1.1学习算法 1.2基本术语和概念 1.3机器学习分类常用算法 1.3.1线性回归 1.3.2逻辑回归 1.3.3决策树 1.3.4朴素贝叶斯 1.3.5支持向量机SVM 1.3.6K-最近临邻KNN 还有K-均值(k-means)、随机森林、降维、人工神经网络等 1.4超参数和验证集 1.4.…

晨持绪科技:抖音网店怎么做有前景

在数字时代的浪潮中,抖音平台以其独特的魅力和庞大的用户基础成为电商的新阵地。开设一家有前景的抖音网店,不仅需要对市场脉搏有敏锐的洞察力,还需融合创新思维与数据驱动的营销策略。 明确定位是成功的先声。深入分析目标消费群体的需求与偏…