大数据-玩转数据-双流JOIN
一、双流JOIN
在Flink中, 支持两种方式的流的Join: Window Join和Interval Join
二、Window Join
窗口join会join具有相同的key并且处于同一个窗口中的两个流的元素.
注意:
1.所有的窗口join都是 inner join, 意味着a流中的元素如果在b流中没有对应的, 则a流中这个元素就不会处理(就是忽略掉了)
2.join成功后的元素的会以所在窗口的最大时间作为其时间戳. 例如窗口[5,10), 则元素会以9作为自己的时间戳。
Window join 仍然可分为 滚动窗口、滑动窗口Join、会话窗口Join
滚动窗口Join代码段示例

package com.lyh.flink12;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;/*** @Author lizhenchao@atguigu.cn* @Date 2021/1/24 22:09*/
public class Flink01_Join_Window_Tumbling {public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());env.setParallelism(1);SingleOutputStreamOperator<WaterSensor> s1 = env.socketTextStream("hadoop100", 8888) // 在socket终端只输入毫秒级别的时间戳.map(value -> {String[] datas = value.split(",");return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));}).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {@Overridepublic long extractTimestamp(WaterSensor element, long recordTimestamp) {return element.getTs() * 1000;}}));SingleOutputStreamOperator<WaterSensor> s2 = env.socketTextStream("hadoop100", 9999) // 在socket终端只输入毫秒级别的时间戳.map(value -> {String[] datas = value.split(",");return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));}).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {@Overridepublic long extractTimestamp(WaterSensor element, long recordTimestamp) {return element.getTs() * 1000;}}));s1.join(s2).where(WaterSensor::getId).equalTo(WaterSensor::getId).window(TumblingEventTimeWindows.of(Time.seconds(5))) // 必须使用窗口.apply(new JoinFunction<WaterSensor, WaterSensor, String>() {@Overridepublic String join(WaterSensor first, WaterSensor second) throws Exception {return "first: " + first + ", second: " + second;}}).print();try {env.execute();} catch (Exception e) {e.printStackTrace();}}
}
运行结果:

三、Interval Join
间隔流join(Interval Join), 是指使用一个流的数据按照key去join另外一条流的指定范围的数据.
如下图: 橙色的流去join绿色的流.范围是由橙色流的event-time + lower bound和event-time + upper bound来决定的.
orangeElem.ts + lowerBound <= greenElem.ts <= orangeElem.ts + upperBound

Interval Join只支持event-time
必须是keyBy之后的流才可以interval join
package com.lyh.flink12;
import com.lyh.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.planner.expressions.In;
import org.apache.flink.util.Collector;
import java.time.Duration;public class Sql_Join_Windows_Interval{public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());env.setParallelism(2);SingleOutputStreamOperator<WaterSensor> s1 = env.socketTextStream("hadoop100", 8888).map(value -> {String[] data = value.split(",");return new WaterSensor(data[0],Long.valueOf(data[1]),Integer.valueOf(data[2]));}).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {@Overridepublic long extractTimestamp(WaterSensor element, long timestamp) {return element.getTs();}}));SingleOutputStreamOperator<WaterSensor> s2 = env.socketTextStream("hadoop100", 9999).map(value -> {String[] data = value.split(",");return new WaterSensor(data[0],Long.valueOf(data[1]),Integer.valueOf(data[2]));}).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {@Overridepublic long extractTimestamp(WaterSensor element, long timestamp) {return element.getTs();}}));s1.keyBy(WaterSensor::getId).intervalJoin(s2.keyBy(WaterSensor::getId)).between(Time.seconds(-2),Time.seconds(3)).process(new ProcessJoinFunction<WaterSensor, WaterSensor, String>() {@Overridepublic void processElement(WaterSensor left,WaterSensor right,Context ctx,Collector<String> out) throws Exception {out.collect(left + "," + right);}}).print();try{env.execute();} catch (Exception e){e.printStackTrace();}}}
运行结果:

相关文章:
大数据-玩转数据-双流JOIN
一、双流JOIN 在Flink中, 支持两种方式的流的Join: Window Join和Interval Join 二、Window Join 窗口join会join具有相同的key并且处于同一个窗口中的两个流的元素. 注意: 1.所有的窗口join都是 inner join, 意味着a流中的元素如果在b流中没有对应的, 则a流中这个元素就不会…...
from PIL import Image,文字成图,ImageFont import jieba分词,input优雅python绘制图片
开始的代码 import os from PIL import Image, ImageDraw, ImageFont import jiebadef generate_image_with_white_bg(text, font_path, output_path):# 设置图片大小和背景颜色image_width 800image_height 600bg_color (255, 255, 255) # 白色# 创建图片对象image Imag…...
渗透测试信息收集方法笔记
一、指纹识别 1、钟馗之眼https://www.zoomeye.org/ 2、天眼查https://www.tianyancha.com/ 3、工具:御剑WEB指纹识别系统正式版,可以查网站用了哪些框架,什么版本,有哪些漏洞 4、kali whatweb 二、信息泄露 1、csdn https://www.…...
协议栈——连接服务器
如对方的ip和port配置信息,这里的连接是指通信前的准备工作 上一篇介绍查看套接字的命令时,可以看到很多信息,但是刚刚创建出来的套接字是什么信息都没有的,协议栈也因此不知道和谁通信; 客户端填补信息 这一步中调…...
数据结构--队列与循环队列的实现
数据结构–队列的实现 1.队列的定义 比如有一个人叫做张三,这天他要去医院看病,看病时就需要先挂号,由于他来的比较晚,所以他的号码就比较大,来的比较早的号码就比较小,需要到就诊窗口从小号到大依次排队,前面的小号就诊结束之后,才会轮到大号来,小号每就诊完毕就销毁,每新来…...
数据结构—栈、队列、链表
一、栈 Stack(存取O(1)) 先进后出,进去123,出来321。 基于数组:最后一位为栈尾,用于取操作。 基于链表:第一位为栈尾,用于取操作。 1.1、数组栈 /*** 基于数组实现的顺序栈&#…...
2023年4月到7月工作经历
2023年4 有同事说程序崩溃一起分析得结果 unsigned uNum 2; std::string str "abc" uNum; std::cout << str; 结果是c 。如果uNum 很大的话,就可能崩溃。 unsigned uNum 2; //std::string str "abc" uN…...
嵌入式Linux应用开发-驱动大全-同步与互斥③
嵌入式Linux应用开发-驱动大全-同步与互斥③ 第一章 同步与互斥③1.4 Linux锁的介绍与使用1.4.1 锁的类型1.4.1.1 自旋锁1.4.1.2 睡眠锁 1.4.2 锁的内核函数1.4.2.1 自旋锁1.4.2.2 信号量1.4.2.3 互斥量1.4.2.4 semaphore和 mutex的区别 1.4.3 何时用何种锁1.4.4 内核抢占(pree…...
力扣-383.赎金信
Idea 使用一个hashmap 或者一个int数组存储第二次字符串中每一个字符及其出现的次数 遍历第一个字符串,讲出现的重复字符减1,若该字符次数已经为0,则返回false AC Code class Solution { public:bool canConstruct(string ransomNote, strin…...
计算机网络 第二章物理层
计算机网络第二章知识点速刷 其中重要的是信源和信宿,以及调制解调器在通信模型当中起到的作用。...
uniapp:动态修改页面标题
我们经常遇到这种情况,点击新增按钮,进入一个空白表单页面,点击修改按钮,其实也是进入这个表单页面,只是表单内容已经被数据库的记录反显了,为了区别页面,我们还需要动态设置页面的标题…...
java学生管理系统
一、项目概述 本学生管理系统旨在提供一个方便的界面,用于学校或机构管理学生信息,包括学生基本信息、课程成绩等。 二、系统架构 系统采用经典的三层架构,包括前端使用JavaSwing,后端采用Java Servlet,数据库使用M…...
Docker和容器化:简介和使用案例
Docker和容器化:简介和使用案例 引言 容器化技术在近年来变得越来越流行,为开发人员和运维团队提供了更加灵活、高效的软件部署和管理方式。其中,Docker是最为知名和广泛使用的容器化平台之一。本篇博客文章将介绍Docker和容器化的基本概念…...
(高阶) Redis 7 第18讲 RedLock 分布式锁
🌹 以下分享 RedLock 分布式锁,如有问题请指教。🌹🌹 如你对技术也感兴趣,欢迎交流。🌹🌹🌹 如有对阁下帮助,请👍点赞💖收藏🐱🏍分享😀 问题 分布式锁问题从(高阶) Redis 7 第17讲 分布式锁 实战篇_PJ码匠人的博客-CSDN博客 这篇文章来看,…...
嵌入式软件架构基础设施设计方法
大家好,今天分享一篇嵌入式软件架构设计相关的文章。 软件架构这东西,众说纷纭,各有观点。在我看来,软件架构是软件系统的基本结构,包含其组件、组件之间的关系、组件设计与演进的规则,以及体现这些规则的基…...
MySQL进阶_3.性能分析工具的使用
文章目录 第一节、数据库服务器的优化步骤第二节、查看系统性能参数第三节、 慢查询日志第四节、 查看 SQL 执行成本第五节、 分析查询语句:EXPLAIN5.1 基本语法5.2 EXPLAIN各列作用 第一节、数据库服务器的优化步骤 当我们遇到数据库调优问题的时候,可…...
Scala第十三章节
Scala第十三章节 1. 高阶函数介绍 2. 作为值的函数 3. 匿名函数 4. 柯里化 5. 闭包 6. 控制抽象 7. 案例: 计算器 scala总目录 文档资料下载...
Nginx高级 第一部分:扩容
Nginx高级 第一部分:扩容 通过扩容提升整体吞吐量 1.单机垂直扩容:硬件资源增加 云服务资源增加 整机:IBM、浪潮、DELL、HP等 CPU/主板:更新到主流 网卡:10G/40G网卡 磁盘:SAS(SCSI) HDD(机械…...
vue项目上线后去除控制台所有console.log打印-配置说明
方式一 npm i babel-plugin-transform-remove-console --save-dev babel.config.js文件中添加 // 然后在babel.config.js中添加判断 const prodPlugin []if (process.env.NODE_ENV production) { // 如果是生产环境,则自动清理掉打印的日志,但保留…...
《XSS-Labs》02. Level 11~20
XSS-Labs 索引Level-11题解 Level-12题解 Level-13题解 Level-14题解 Level-15题解 Level-16题解 Level-17题解 Level-18~20题解 靶场部署在 VMware - Win7。 靶场地址:https://github.com/do0dl3/xss-labs 只要手动注入恶意 JavaScript 脚本成功,就可以…...
KubeSphere 容器平台高可用:环境搭建与可视化操作指南
Linux_k8s篇 欢迎来到Linux的世界,看笔记好好学多敲多打,每个人都是大神! 题目:KubeSphere 容器平台高可用:环境搭建与可视化操作指南 版本号: 1.0,0 作者: 老王要学习 日期: 2025.06.05 适用环境: Ubuntu22 文档说…...
龙虎榜——20250610
上证指数放量收阴线,个股多数下跌,盘中受消息影响大幅波动。 深证指数放量收阴线形成顶分型,指数短线有调整的需求,大概需要一两天。 2025年6月10日龙虎榜行业方向分析 1. 金融科技 代表标的:御银股份、雄帝科技 驱动…...
Unity3D中Gfx.WaitForPresent优化方案
前言 在Unity中,Gfx.WaitForPresent占用CPU过高通常表示主线程在等待GPU完成渲染(即CPU被阻塞),这表明存在GPU瓶颈或垂直同步/帧率设置问题。以下是系统的优化方案: 对惹,这里有一个游戏开发交流小组&…...
智慧工地云平台源码,基于微服务架构+Java+Spring Cloud +UniApp +MySql
智慧工地管理云平台系统,智慧工地全套源码,java版智慧工地源码,支持PC端、大屏端、移动端。 智慧工地聚焦建筑行业的市场需求,提供“平台网络终端”的整体解决方案,提供劳务管理、视频管理、智能监测、绿色施工、安全管…...
PPT|230页| 制造集团企业供应链端到端的数字化解决方案:从需求到结算的全链路业务闭环构建
制造业采购供应链管理是企业运营的核心环节,供应链协同管理在供应链上下游企业之间建立紧密的合作关系,通过信息共享、资源整合、业务协同等方式,实现供应链的全面管理和优化,提高供应链的效率和透明度,降低供应链的成…...
Swift 协议扩展精进之路:解决 CoreData 托管实体子类的类型不匹配问题(下)
概述 在 Swift 开发语言中,各位秃头小码农们可以充分利用语法本身所带来的便利去劈荆斩棘。我们还可以恣意利用泛型、协议关联类型和协议扩展来进一步简化和优化我们复杂的代码需求。 不过,在涉及到多个子类派生于基类进行多态模拟的场景下,…...
【SpringBoot】100、SpringBoot中使用自定义注解+AOP实现参数自动解密
在实际项目中,用户注册、登录、修改密码等操作,都涉及到参数传输安全问题。所以我们需要在前端对账户、密码等敏感信息加密传输,在后端接收到数据后能自动解密。 1、引入依赖 <dependency><groupId>org.springframework.boot</groupId><artifactId...
Linux相关概念和易错知识点(42)(TCP的连接管理、可靠性、面临复杂网络的处理)
目录 1.TCP的连接管理机制(1)三次握手①握手过程②对握手过程的理解 (2)四次挥手(3)握手和挥手的触发(4)状态切换①挥手过程中状态的切换②握手过程中状态的切换 2.TCP的可靠性&…...
【Java_EE】Spring MVC
目录 Spring Web MVC 编辑注解 RestController RequestMapping RequestParam RequestParam RequestBody PathVariable RequestPart 参数传递 注意事项 编辑参数重命名 RequestParam 编辑编辑传递集合 RequestParam 传递JSON数据 编辑RequestBody …...
【RockeMQ】第2节|RocketMQ快速实战以及核⼼概念详解(二)
升级Dledger高可用集群 一、主从架构的不足与Dledger的定位 主从架构缺陷 数据备份依赖Slave节点,但无自动故障转移能力,Master宕机后需人工切换,期间消息可能无法读取。Slave仅存储数据,无法主动升级为Master响应请求ÿ…...
