电子相册在线制作网站,东莞网站建设开发,网站建设免费模板哪家好,建设网站建设多少钱背景
之前我们在一片文章里简单介绍过Flink的多流合并算子
java Flink#xff08;三十六#xff09;Flink多流合并算子UNION、CONNECT、CoGroup、Join
今天我们通过Flink 1.14的源码对Flink的Interval Join进行深入的理解。
Interval Join不是两个窗口做关联#xff0c;…背景
之前我们在一片文章里简单介绍过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类下的方法 包装返回类型的TypeInfomationTypeInfo的介绍可以看上一篇 返回的outputType SingleOutputStreamOperator使用给定的用户函数完成联接操作该函数针对每个联接的元素对执行。这种方法允许传递输出类型的显式类型信息。
IntervalJoinOperator初始化
左边界右边界检查 获取左流还有右流数据对应的序列化从TypeInfo获取的 继续看IntervalJoinOperator中的其余关键实现
open方法用来注册定时器
初始化两个流的map状态
处理左侧流中的数据。每当数据到达左流时它就会被添加到左缓冲区。将从右侧缓冲区中查找该元素可能的候选联接如果该对位于用户定义的边界内则将其传递给 ProcessJoinFunction
同理处理右流
进入数据处理函数
获取数据取出事件时间
超过当前watermark的数据进行过滤
数据没问题的话将数据添加到状态
遍历另一条流的状态遍历其中的数据把满足时间要求的数据进行collect 注册一个当前事件时间戳右边界的定时器
定时器触发后清空map状态中时间戳-左边界的那条数据
简单实例
pom
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdorg.example/groupIdartifactIdFlinkCode/artifactIdversion1.0-SNAPSHOT/versionpropertiesmaven.compiler.source8/maven.compiler.sourcemaven.compiler.target8/maven.compiler.targetjdk.version1.8/jdk.versionjar.nameubs-data-converter/jar.nameproject.build.sourceEncodingUTF-8/project.build.sourceEncoding!--Flink 版本--flink.version1.14.4/flink.version/propertiesdependenciesdependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-java_2.11/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-kafka_2.11/artifactIdversion${flink.version}/versionexclusionsexclusiongroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactId/exclusion/exclusions/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-runtime-web_2.11/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.httpcomponents/groupIdartifactIdhttpclient/artifactIdversion4.5.10/version/dependencydependencygroupIdcom.alibaba/groupIdartifactIdfastjson/artifactIdversion1.2.8/version/dependencydependencygroupIdorg.apache.avro/groupIdartifactIdavro/artifactIdversion1.9.2/version/dependencydependencygroupIdorg.apache.httpcomponents/groupIdartifactIdhttpcore/artifactIdversion4.4.1/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.18.16/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion1.18.16/versionscopecompile/scope/dependency/dependenciesbuildpluginsplugingroupIdorg.apache.avro/groupIdartifactIdavro-maven-plugin/artifactIdversion1.9.2/versionexecutionsexecutionphasegenerate-sources/phasegoalsgoalschema/goal/goalsconfigurationsourceDirectory${project.basedir}/src/main/resources//sourceDirectory/configuration/execution/executions/pluginplugingroupIdorg.apache.maven.plugins/groupIdartifactIdmaven-compiler-plugin/artifactIdversion3.1/versionconfigurationsource${jdk.version}/sourcetarget${jdk.version}/targetencoding${project.build.sourceEncoding}/encoding/configuration/pluginplugingroupIdorg.apache.maven.plugins/groupIdartifactIdmaven-shade-plugin/artifactIdversion3.1.1/versionexecutionsexecutionphasepackage/phasegoalsgoalshade/goal/goalsconfigurationfinalName${jar.name}/finalNameartifactSetexcludesexcludecom.google.code.findbugs:jsr305/excludeexcludeorg.slf4j:*/excludeexcludelog4j:*/excludeexcludeorg.glassfish.jersey.core:jersey-common/exclude/excludes/artifactSetrelocationsrelocationpatterncom.google.common/patternshadedPatterncom.shade.google.common/shadedPattern/relocationrelocationpatternorg.apache.kafka/patternshadedPatternorg.shade.apache.kafka/shadedPattern/relocation/relocationsfiltersfilterartifact*/artifactincludesincludeorg/apache/htrace/**/includeincludeorg/apache/avro/**/includeincludeorg/apache/flink/streaming/**/includeincludeorg/apache/flink/connector/**/includeincludeorg/apache/kafka/**/includeincludeorg/apache/hive/**/includeincludeorg/apache/hadoop/hive/**/includeincludeorg/apache/curator/**/includeincludeorg/apache/zookeeper/**/includeincludeorg/apache/jute/**/includeincludeorg/apache/thrift/**/includeincludeorg/apache/http/**/includeincludeorg/I0Itec/**/includeincludejline/**/includeincludecom/yammer/**/includeincludekafka/**/includeincludeorg/apache/hadoop/hbase/**/includeincludecom/alibaba/fastjson/**/includeincludeorg/elasticsearch/action/**/includeincludeio/confluent/**/includeincludecom/fasterxml/**/includeincludeorg/elasticsearch/**/includeincludehbase-default.xml/includeincludehbase-site.xml/include/includes/filterfilterartifactorg.apache.hadoop.hive.*:*/artifactexcludesexclude/excludeexclude/excludeexclude/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();//设置watermarkWatermarkStrategyUser userWatermarkStrategy WatermarkStrategy.UserforBoundedOutOfOrderness(Duration.ofSeconds(1)).withTimestampAssigner(new SerializableTimestampAssignerUser() {Overridepublic long extractTimestamp(User element, long recordTimestamp) {return element.getT();}});DataStreamUser userDataStreamSource env.addSource(new UserSource()).assignTimestampsAndWatermarks(userWatermarkStrategy);//设置watermarkWatermarkStrategyOrder orderWatermarkStrategy WatermarkStrategy.OrderforBoundedOutOfOrderness(Duration.ofSeconds(1)).withTimestampAssigner(new SerializableTimestampAssignerOrder() {Overridepublic long extractTimestamp(Order element, long recordTimestamp) {return element.getTime();}});DataStreamOrder orderDataStreamSource env.addSource(new OrderSource()).assignTimestampsAndWatermarks(orderWatermarkStrategy);env.setParallelism(1);SingleOutputStreamOperatorString process userDataStreamSource.keyBy(o - o.getId()).intervalJoin(orderDataStreamSource.keyBy(o - o.getId())).between(Time.seconds(-5), Time.seconds(0)).process(new ProcessJoinFunctionUser, Order, String() {Overridepublic void processElement(User left, Order right, ProcessJoinFunctionUser, Order, String.Context ctx, CollectorString 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();}
}
