java Flink（四十三）Flink Interval Join源码解析以及简单实例

背景

之前我们在一片文章里简单介绍过Flink的多流合并算子

java Flink（三十六）Flink多流合并算子UNION、CONNECT、CoGroup、Join

今天我们通过Flink 1.14的源码对Flink的Interval Join进行深入的理解。

Interval Join不是两个窗口做关联，更适用于处理乱序数据流之间的关联。它的作用更类似于从左流中a元素本身出发，对右流中一段时间内的数据进行关联（Inner Join：只关联相同Key的数据）。

如图所示：

 下边这条流中的2关联到上范围内的0/1

​

源码解析

Flink版本1.14.4

按住Ctrl+鼠标左键，点击process进入源码

​

 这里process方法是在KeydStream.java下IntervalJoined类下的方法

 包装返回类型的TypeInfomation（TypeInfo的介绍可以看上一篇）

 返回的outputType

SingleOutputStreamOperator使用给定的用户函数完成联接操作，该函数针对每个联接的元素对执行。这种方法允许传递输出类型的显式类型信息。

​

 IntervalJoinOperator初始化

​​

左边界<=右边界检查

获取左流还有右流数据对应的序列化（从TypeInfo获取的）

继续看IntervalJoinOperator中的其余关键实现

open方法用来注册定时器

​

 初始化两个流的map状态

​

处理左侧流中的数据。每当数据到达左流时，它就会被添加到左缓冲区。将从右侧缓冲区中查找该元素可能的候选联接，如果该对位于用户定义的边界内，则将其传递给 ProcessJoinFunction

​ 同理处理右流​

进入数据处理函数

获取数据，取出事件时间

​

 超过当前watermark的数据进行过滤

 ​

数据没问题的话，将数据添加到状态

 ​​

遍历另一条流的状态，遍历其中的数据，把满足时间要求的数据进行collect

​注册一个当前事件时间戳+右边界的定时器

​​

定时器触发后，清空map状态中时间戳-左边界的那条数据​

简单实例 

pom

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>FlinkCode</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <jdk.version>1.8</jdk.version>
        <jar.name>ubs-data-converter</jar.name>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <!--Flink 版本-->
        <flink.version>1.14.4</flink.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_2.11</artifactId>
            <version>${flink.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka-clients</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web_2.11</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.httpcomponents</groupId>
            <artifactId>httpclient</artifactId>
            <version>4.5.10</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.8</version>
        </dependency>
        <dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro</artifactId>
            <version>1.9.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.httpcomponents</groupId>
            <artifactId>httpcore</artifactId>
            <version>4.4.1</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.16</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.16</version>
            <scope>compile</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.avro</groupId>
                <artifactId>avro-maven-plugin</artifactId>
                <version>1.9.2</version>
                <executions>
                    <execution>
                        <phase>generate-sources</phase>
                        <goals>
                            <goal>schema</goal>
                        </goals>
                        <configuration>
                            <sourceDirectory>${project.basedir}/src/main/resources/</sourceDirectory>
                           
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>${jdk.version}</source>
                    <target>${jdk.version}</target>
                    <encoding>${project.build.sourceEncoding}</encoding>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.1.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <finalName>${jar.name}</finalName>
                            <artifactSet>
                                <excludes>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                    <exclude>log4j:*</exclude>
                                    <exclude>org.glassfish.jersey.core:jersey-common</exclude>
                                </excludes>
                            </artifactSet>
                            <relocations>
                                <relocation>
                                    <pattern>com.google.common</pattern>
                                    <shadedPattern>com.shade.google.common</shadedPattern>
                                </relocation>
                                <relocation>
                                    <pattern>org.apache.kafka</pattern>
                                    <shadedPattern>org.shade.apache.kafka</shadedPattern>
                                </relocation>
                            </relocations>
                            <filters>
                                <filter>
                                    <artifact>*</artifact>
                                    <includes>
                                        <include>org/apache/htrace/**</include>
                                        <include>org/apache/avro/**</include>
                                        <include>org/apache/flink/streaming/**</include>
                                        <include>org/apache/flink/connector/**</include>
                                        <include>org/apache/kafka/**</include>
                                        <include>org/apache/hive/**</include>
                                        <include>org/apache/hadoop/hive/**</include>
                                        <include>org/apache/curator/**</include>
                                        <include>org/apache/zookeeper/**</include>
                                        <include>org/apache/jute/**</include>
                                        <include>org/apache/thrift/**</include>
                                        <include>org/apache/http/**</include>
                                        <include>org/I0Itec/**</include>
                                        <include>jline/**</include>
                                        <include>com/yammer/**</include>
                                        <include>kafka/**</include>
                                        <include>org/apache/hadoop/hbase/**</include>
                                        <include>com/alibaba/fastjson/**</include>
                                        <include>org/elasticsearch/action/**</include>
                                        <include>io/confluent/**</include>
                                        <include>com/fasterxml/**</include>
                                        <include>org/elasticsearch/**</include>
                                        <include>hbase-default.xml</include>
                                        <include>hbase-site.xml</include>
                                    </includes>
                                </filter>
                                <filter>
                                    <artifact>org.apache.hadoop.hive.*:*</artifact>
                                    <excludes>
                                        <exclude></exclude>
                                        <exclude></exclude>
                                        <exclude></exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

user bean

package ubs.app.intervaljoin.bean;

import lombok.*;

@Data
@AllArgsConstructor
@Setter
@Getter
@NoArgsConstructor
public class User{
    Integer id;

    Long t;
    
}

 order bean 

package ubs.app.intervaljoin.bean;

import lombok.*;

@Data
@AllArgsConstructor
@Setter
@Getter
@NoArgsConstructor
public class Order {

    Integer id;

    Long price;

    Long time;

}

main

package ubs.app.intervaljoin;


import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.datastream.DataStream;
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.util.Collector;
import ubs.app.intervaljoin.bean.Order;
import ubs.app.intervaljoin.bean.User;
import ubs.app.intervaljoin.source.OrderSource;
import ubs.app.intervaljoin.source.UserSource;

import java.time.Duration;

public class IntervalJoinApp  {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置watermark
        WatermarkStrategy<User> userWatermarkStrategy = WatermarkStrategy.<User>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                .withTimestampAssigner(new SerializableTimestampAssigner<User>() {

                    @Override
                    public long extractTimestamp(User element, long recordTimestamp) {
                        return element.getT();
                    }
                });

        DataStream<User> userDataStreamSource = env.addSource(new UserSource()).assignTimestampsAndWatermarks(userWatermarkStrategy);
        //设置watermark
        WatermarkStrategy<Order> orderWatermarkStrategy = WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                .withTimestampAssigner(new SerializableTimestampAssigner<Order>() {
                    @Override
                    public long extractTimestamp(Order element, long recordTimestamp) {
                        return element.getTime();
                    }
                });
        DataStream<Order> orderDataStreamSource = env.addSource(new OrderSource()).assignTimestampsAndWatermarks(orderWatermarkStrategy);

        env.setParallelism(1);
        SingleOutputStreamOperator<String> process = userDataStreamSource.keyBy(o -> o.getId())
                .intervalJoin(orderDataStreamSource.keyBy(o -> o.getId()))
                .between(Time.seconds(-5), Time.seconds(0))
                .process(
                        new ProcessJoinFunction<User, Order, String>() {

                            @Override
                            public void processElement(User left, Order right, ProcessJoinFunction<User, Order, String>.Context ctx, Collector<String> out) throws Exception {
                                Integer lid = left.getId();
                                Long lt = left.getT();
                                Integer rid = right.getId();
                                long rt = right.getTime();
                                out.collect(String.format("左%s 左时间%s 右%s 右时间%s 关联到了 %s", lid, lt/1000, rid, rt/1000, rt/1000-lt/1000));
                            }
                        }
                );
        process.print();
        env.execute();
    }
}

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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

001_【基础篇】SpringBoot入门案例创建与实现

001_【基础篇】SpringBoot入门案例创建与实现

要求&#xff1a;使用 Springboot 开发一个 web 程序&#xff0c;浏览器发起请求/hello后&#xff0c;给浏览器返回字符串 hello springboot 使用 springboot 只需要引入一个起步依赖 <dependency><groupId>org.springframework.boot</groupId><artifac…
阅读更多...
STP环路避免实验（思科）

STP环路避免实验（思科）

华为设备参考&#xff1a;STP环路避免实验&#xff08;华为&#xff09; 一&#xff0c;技术简介 Spanning Tree Protocol&#xff08;STP&#xff09;&#xff0c;即生成树协议&#xff0c;是一种数据链路层协议。主要作用是防止二层环路&#xff0c;并自适应网络变化和故障…
阅读更多...
个人网站制作 Part 9 添加发布、管理博客功能 | Web开发项目

个人网站制作 Part 9 添加发布、管理博客功能 | Web开发项目

文章目录 &#x1f469;‍&#x1f4bb; 基础Web开发练手项目系列&#xff1a;个人网站制作&#x1f680; 添加博客功能&#x1f528;使用Express和MongoDB&#x1f527;步骤 1: 创建博客模型&#x1f527;步骤 2: 创建博客路由 &#x1f528;使用前端框架&#x1f527;步骤 3:…
阅读更多...
什么是零日攻击？

什么是零日攻击？

一、零日攻击的概念 零日攻击是指利用零日漏洞对系统或软件应用发动的网络攻击。 零日漏洞也称零时差漏洞&#xff0c;通常是指还没有补丁的安全漏洞。由于零日漏洞的严重级别通常较高&#xff0c;所以零日攻击往往也具有很大的破坏性。 目前&#xff0c;任何安全产品或解决方案…
阅读更多...
OxyPlot 导出图片

OxyPlot 导出图片

在 OxyPlot 官方文档 https://oxyplot.readthedocs.io/en/latest/export/index.html 中查看 这里用到的是导出到 PNG 文件的方法&#xff0c;不过用的 NuGet 包最新版&#xff08;2.1.0&#xff09;中&#xff0c;PngExporter 中并没有 Background 属性&#xff1a; 所以如果图…
阅读更多...
字符函数与字符串函数

字符函数与字符串函数

前言 本次博客可以说内容最为多的一次博客&#xff0c;讲解同样很细致大家好好看看 1字符函数 在讲解字符函数时,大家得了解什么是字符吧 普通字符a b c 1 转义字符 \n 换行‘ \t’ 水平制表符\r回车 大家了解即可 在C语言中字符也可以有分类 所以我们先来看看…
阅读更多...
软件测试经验与教训

软件测试经验与教训

大概在18年的时候&#xff0c;就看过《软件测试经验与教训》的纸制版&#xff0c;里面的一些观点深刻的影响了我&#xff0c;也影响了后来我对测试的思考。最近又一次快速阅读了电子版&#xff0c;还是收获满满。下面精选出10条&#xff0c;和大家分享。 一、测试人员是项目的…
阅读更多...
testng测试类第2步

testng测试类第2步

创建xml文件并编写xml文件 并学习其中的参数 1、创建 xml文件 在测试包->右键找到creat TestNG XML 创建xml文件 如果报错就可以粘贴过来 认识原始文件 这里首行是标识&#xff0c;其次是2个参数&#xff0c;name是测试套件的名称&#xff0c;谁的测试套件一般是公司名称…
阅读更多...
JAVA实战手册-开篇总述

JAVA实战手册-开篇总述

该专题以实战为出发点&#xff0c;总结概述了实际工作中常用的java知识点&#xff0c;掌握了这些知识点&#xff0c;日常工作开发以及面试都不在话下。 话不多说&#xff0c;直入正题&#xff0c;以下为JAVA知识点概括总结&#xff08;总计涵盖了10大类78小项&#xff09; 针对…
阅读更多...
AcWing 727. 菱形——像拼图一样做题

AcWing 727. 菱形——像拼图一样做题

题目描述] 分析&#xff1a; 利用程序根据输入的整数&#xff0c;画出由字符*构成的该整数阶的实心菱形。给出一个示例&#xff1a; n 7 n7 n7。 * * * * * * * * * * * * * * * * * * * * * * * * * 我们将采取拆解问题&#xff0c;通过四个部分的…
阅读更多...
Linux编程4.7 网络编程-套接字与地址

Linux编程4.7 网络编程-套接字与地址

1、因特网地址结构 struct in_addr{in_addr_t s_addr; /*ipv4地址*/ }struct sockaaddr_in{short int sin_family; /*Internet地址族&#xff0c;如AF_INET&#xff08;主机字节序&#xff09;*/ unsigned short int sin_port; /*端口号&#xff0c;16…
阅读更多...
upload-labs第二关第五关

upload-labs第二关第五关

upload-labs第二关 1、先上传个马试试&#xff0c;看看页面显示什么&#xff0c;要先把上次上传的马删掉哦 2、提示文件类型不正确&#xff0c;在看下提示说是在服务端做了检查 3、那就上传个php格式的&#xff0c;然后呢bp抓包把文件格式改了 4、修改后的 5、修改后发现不…
阅读更多...
【物联网】Modbus 协议及应用

【物联网】Modbus 协议及应用

Modbus 协议简介 QingHub设计器在设计物联网数据采集时不可避免的需要针对Modbus协议的设备做相关数据采集&#xff0c;这里就我们的实际项目经验分享Modbus协议 简介 Modbus由MODICON公司于1979年开发&#xff0c;是一种工业现场总线协议标准。1996年施耐德公司推出基于以太…
阅读更多...
ansible 运维自动化

ansible 运维自动化

pxe 一键安装操作系统 操作系统只是提供一个平台 lnmp 需要多软件协同完成的一个简单项目 服务器正常运行 日常运维 巡检 服务器上的软件正常运行 zabbix 普罗米修斯 系统调优&#xff0c;架构调优 前言 运维自动化 云计算核心职能 搭建平台架构 …
阅读更多...
TCPIP协议总结

TCPIP协议总结

一、TCP的三次握手 TCP连接的建立时&#xff0c;双方需要经过三次握手&#xff0c;而断开连接时&#xff0c;双方需要经过四次分手&#xff0c;那么&#xff0c;其三次握手和四次分手分别做了什么呢&#xff1f;又是如何进行的呢&#xff1f; 通常情况下&#xff0c;建立连接的…
阅读更多...
从零开始学习如何使用 Postman 请求头

从零开始学习如何使用 Postman 请求头

当你在使用 Postman 发送请求时&#xff0c;请求头&#xff08;Headers&#xff09;是你可以包含在 HTTP 请求中的重要部分之一。请求头包含了关于请求的元数据信息&#xff0c;这些信息对于服务器来处理请求是非常重要的。下面是一份详细的图文介绍&#xff0c;说明了如何在 P…
阅读更多...
电商数据采集效率开挂【Python电商数据采集API接口】

电商数据采集效率开挂【Python电商数据采集API接口】

数据监测 监测线上电商平台的商品、店铺数据&#xff0c;包括商品销量/库存/价格/店铺等级/发货地/促销活动等信息&#xff0c;支持十多个国内主流电商平台。 在线维权 实现多平台在线一键投诉&#xff0c;与各大电商投诉平台系统对接&#xff0c;实时同步投诉进展。 渠道管…
阅读更多...
C# visual studio 2022 学习2

C# visual studio 2022 学习2

类成员&#xff1a; 1.字段成员 字段只是类中声明的一个变量&#xff0c;用来在对象中存储信息。 &#xff08;1&#xff09;.静态字段 使用static关键字修饰的就是静态字段&#xff0c;静态字段属于类而不属于某个类实例&#xff0c;对它的访问使用“类名.静态字段名” &…
阅读更多...
28-2 文件上传漏洞 - 双写绕过

28-2 文件上传漏洞 - 双写绕过

环境准备&#xff1a;构建完善的安全渗透测试环境&#xff1a;推荐工具、资源和下载链接_渗透测试靶机下载-CSDN博客 一、双写绕过原理 在代码编写过程中&#xff0c;双写绕过原理指的是只对黑名单中的内容进行一次空替换。由于只进行一次替换&#xff0c;导致了双写绕过的情况…
阅读更多...
数据结构从入门到精通——排序的概念及运用

数据结构从入门到精通——排序的概念及运用

排序的概念及运用 前言一、排序的概念排序稳定性内部排序外部排序 二、排序运用三、常见的排序算法四、排序性能检测代码srand()clock() 五、oj排序测试代码 前言 排序是将数据按照一定规则重新排列的过程&#xff0c;常见规则有升序、降序等。排序算法如冒泡排序、快速排序等…
阅读更多...
最新文章

Copyright @ 2022~2023