二次开发Flink-coGroup算子支持迟到数据通过测输出流提取

1.背景

coGroup算子开窗到时间关闭之后，迟到数据无法通过测输出流提取，intervalJoin算子提供了api，因为join算子底层就是coGroup算子，所以Join算子也不行。

flink版本 v1.17.1

2.coGroup算子源码分析

2.1完成的coGroup算子调用流程

    input1.coGroup(input2).where(keySelector1).equalTo(keySelector2).window(windowAssigner).trigger(trigger).evictor(evictor).allowedLateness(allowedLateness).apply(cgroupFunction)

通过上述代码可以看到没有sideOutputLateData的相关方法,用来提取窗口关闭之后的迟到数据

2.2coGroup方法入口

其中创建了一个CoGroupedStreams流对象

    /*** Creates a join operation. See {@link CoGroupedStreams} for an example of how the keys and* window can be specified.*/public <T2> CoGroupedStreams<T, T2> coGroup(DataStream<T2> otherStream) {return new CoGroupedStreams<>(this, otherStream);}

2.3 CoGroupedStreams对象分析

他可以理解为构造设计模式的一个Builder类，通过where方法配置第一条流的KeySelector，再返回一个CoGroupedStreams的内部类Where，再通过equalTo方法配置第二条流的KeySelector，再返回EqualTo内部类，window方法配置窗口划分器，返回WithWindow内部类，后续都是窗口的配置 trigger，evictor，allowedLateness配置窗口参数，最后调用apply方法传送用户业务函数

2.4WithWindow内部类分析

WithWindow是最终保存所有配置的内部类包括两条流，窗口配置，key提取器的配置，最终会用户调用apply方法触发CoGroup的业务，在apply方法中通过union联合两条流，然后通过keyby转为KeyedStream，再通过window配置窗口，最终调用窗口函数的apply方法，传入WindowFunction，做CoGroup的业务与用户业务。

具体代码如下已写好备注

/*** A co-group operation that has {@link KeySelector KeySelectors} defined for both inputs as* well as a {@link WindowAssigner}.** @param <T1> Type of the elements from the first input* @param <T2> Type of the elements from the second input* @param <KEY> Type of the key. This must be the same for both inputs* @param <W> Type of {@link Window} on which the co-group operation works.*/@Publicpublic static class WithWindow<T1, T2, KEY, W extends Window> {//第一条流private final DataStream<T1> input1;//第二条流private final DataStream<T2> input2;//第一个key提取器private final KeySelector<T1, KEY> keySelector1;//第二个Key提取器private final KeySelector<T2, KEY> keySelector2;//Key的类型private final TypeInformation<KEY> keyType;//窗口分配器private final WindowAssigner<? super TaggedUnion<T1, T2>, W> windowAssigner;//窗口出发计算器private final Trigger<? super TaggedUnion<T1, T2>, ? super W> trigger;private final Evictor<? super TaggedUnion<T1, T2>, ? super W> evictor;private final Time allowedLateness;private WindowedStream<TaggedUnion<T1, T2>, KEY, W> windowedStream;//构造函数给上面对象赋值protected WithWindow(DataStream<T1> input1,DataStream<T2> input2,KeySelector<T1, KEY> keySelector1,KeySelector<T2, KEY> keySelector2,TypeInformation<KEY> keyType,WindowAssigner<? super TaggedUnion<T1, T2>, W> windowAssigner,Trigger<? super TaggedUnion<T1, T2>, ? super W> trigger,Evictor<? super TaggedUnion<T1, T2>, ? super W> evictor,Time allowedLateness) {this.input1 = input1;this.input2 = input2;this.keySelector1 = keySelector1;this.keySelector2 = keySelector2;this.keyType = keyType;this.windowAssigner = windowAssigner;this.trigger = trigger;this.evictor = evictor;this.allowedLateness = allowedLateness;}/*** Completes the co-group operation with the user function that is executed for windowed* groups.** <p>Note: This method's return type does not support setting an operator-specific* parallelism. Due to binary backwards compatibility, this cannot be altered. Use the* {@link #with(CoGroupFunction, TypeInformation)} method to set an operator-specific* parallelism.*/public <T> DataStream<T> apply(CoGroupFunction<T1, T2, T> function, TypeInformation<T> resultType) {// clean the closurefunction = input1.getExecutionEnvironment().clean(function);//创建合并两个流的公共TypeInfo，UnionTypeInfo最终会将Input1，Input2的数据通过map算子转换为该类型UnionTypeInfo<T1, T2> unionType =new UnionTypeInfo<>(input1.getType(), input2.getType());//转换成union的KeySelectorUnionKeySelector<T1, T2, KEY> unionKeySelector =new UnionKeySelector<>(keySelector1, keySelector2);//将taggedInput1的数据类容map成UnionTypeInfo<T1, T2>类型SingleOutputStreamOperator<TaggedUnion<T1, T2>> taggedInput1 =input1.map(new Input1Tagger<T1, T2>());taggedInput1.getTransformation().setParallelism(input1.getParallelism(), false);taggedInput1.returns(unionType);//将taggedInput2的数据类容map成UnionTypeInfo<T1, T2>类型SingleOutputStreamOperator<TaggedUnion<T1, T2>> taggedInput2 =input2.map(new Input2Tagger<T1, T2>());taggedInput2.getTransformation().setParallelism(input2.getParallelism(), false);taggedInput2.returns(unionType);//将两个流进行unionDataStream<TaggedUnion<T1, T2>> unionStream = taggedInput1.union(taggedInput2);//keyBy并且开窗windowedStream =new KeyedStream<TaggedUnion<T1, T2>, KEY>(unionStream, unionKeySelector, keyType).window(windowAssigner);//配置窗口触发器if (trigger != null) {windowedStream.trigger(trigger);}//配置移除器if (evictor != null) {windowedStream.evictor(evictor);}//配置allowedLatenessif (allowedLateness != null) {windowedStream.allowedLateness(allowedLateness);}//创建CoGroupWindowFunction ，并把用户函数传入进去return windowedStream.apply(new CoGroupWindowFunction<T1, T2, T, KEY, W>(function), resultType);}/*** Completes the co-group operation with the user function that is executed for windowed* groups.** <p><b>Note:</b> This is a temporary workaround while the {@link #apply(CoGroupFunction,* TypeInformation)} method has the wrong return type and hence does not allow one to set an* operator-specific parallelism** @deprecated This method will be removed once the {@link #apply(CoGroupFunction,*     TypeInformation)} method is fixed in the next major version of Flink (2.0).*/@PublicEvolving@Deprecatedpublic <T> SingleOutputStreamOperator<T> with(CoGroupFunction<T1, T2, T> function, TypeInformation<T> resultType) {return (SingleOutputStreamOperator<T>) apply(function, resultType);}@VisibleForTestingTime getAllowedLateness() {return allowedLateness;}//获取窗口包装流，但是标记为VisibleForTesting，用户无法调用，如果可以调用的话可以通过该方法获取包装流之后通过窗口流获取迟到数据的测输出流@VisibleForTestingWindowedStream<TaggedUnion<T1, T2>, KEY, W> getWindowedStream() {return windowedStream;}}

2.5CoGroupWindowFunction函数分析

CoGroupWindowFunction也是CoGroupedStreams内部类，负责做CoGroup的业务，最终将数据封装好转发给用户函数（也就是2.1中apply中的cgroupFunction）

   private static class CoGroupWindowFunction<T1, T2, T, KEY, W extends Window>extends WrappingFunction<CoGroupFunction<T1, T2, T>>implements WindowFunction<TaggedUnion<T1, T2>, T, KEY, W> {private static final long serialVersionUID = 1L;public CoGroupWindowFunction(CoGroupFunction<T1, T2, T> userFunction) {super(userFunction);}@Overridepublic void apply(KEY key, W window, Iterable<TaggedUnion<T1, T2>> values, Collector<T> out)throws Exception {//缓存当前窗口里1号流的数据List<T1> oneValues = new ArrayList<>();//缓存当前窗口里2号流的数据List<T2> twoValues = new ArrayList<>();for (TaggedUnion<T1, T2> val : values) {if (val.isOne()) {oneValues.add(val.getOne());} else {twoValues.add(val.getTwo());}}//传入到用户函数中wrappedFunction.coGroup(oneValues, twoValues, out);}}

3.修改源码支持获取迟到数据测输出流

思路 复制CoGroupedStreams新增一个NewCoGroupedStreams，在WithWindow函数中增加方法sideOutputLateData，让用户传入outputTag，用于提取窗口关闭后的测输出流。

3.1复制CoGroupedStreams

3.2新增WithWindow.sideOutputLateData方法

新增该方法，传入outputTag，下图WithWindow构造方法是3.3新增的

    @PublicEvolvingpublic WithWindow<T1, T2, KEY, W> sideOutputLateData(OutputTag<TaggedUnion<T1, T2>> outputTag) {return new WithWindow<>(input1,input2,keySelector1,keySelector2,keyType,windowAssigner,trigger,evictor,allowedLateness,outputTag);}

3.3新增WithWindow构造方法

新增属性laterDataOutputTag，用来保存构造函数中传入的laterOutputTag

   protected WithWindow(DataStream<T1> input1,DataStream<T2> input2,KeySelector<T1, KEY> keySelector1,KeySelector<T2, KEY> keySelector2,TypeInformation<KEY> keyType,WindowAssigner<? super TaggedUnion<T1, T2>, W> windowAssigner,Trigger<? super TaggedUnion<T1, T2>, ? super W> trigger,Evictor<? super TaggedUnion<T1, T2>, ? super W> evictor,Time allowedLateness,OutputTag<TaggedUnion<T1, T2>> laterOutputTag) {this(input1,input2,keySelector1,keySelector2,keyType,windowAssigner,trigger,evictor,allowedLateness);this.lateDataOutputTag = laterOutputTag;}

3.4修改apply方法

判断lateDataOutputTag 是否为null，如果不为null则调用windowedStream的sideOutputLateData设置迟到数据tag

 /*** Completes the co-group operation with the user function that is executed for windowed* groups.** <p>Note: This method's return type does not support setting an operator-specific* parallelism. Due to binary backwards compatibility, this cannot be altered. Use the* {@link #with(CoGroupFunction, TypeInformation)} method to set an operator-specific* parallelism.*/public <T> DataStream<T> apply(CoGroupFunction<T1, T2, T> function, TypeInformation<T> resultType) {// clean the closurefunction = input1.getExecutionEnvironment().clean(function);UnionTypeInfo<T1, T2> unionType =new UnionTypeInfo<>(input1.getType(), input2.getType());UnionKeySelector<T1, T2, KEY> unionKeySelector =new UnionKeySelector<>(keySelector1, keySelector2);SingleOutputStreamOperator<TaggedUnion<T1, T2>> taggedInput1 =input1.map(new Input1Tagger<T1, T2>());taggedInput1.getTransformation().setParallelism(input1.getParallelism(), false);taggedInput1.returns(unionType);SingleOutputStreamOperator<TaggedUnion<T1, T2>> taggedInput2 =input2.map(new Input2Tagger<T1, T2>());taggedInput2.getTransformation().setParallelism(input2.getParallelism(), false);taggedInput2.returns(unionType);DataStream<TaggedUnion<T1, T2>> unionStream = taggedInput1.union(taggedInput2);// we explicitly create the keyed stream to manually pass the key type information inwindowedStream =new KeyedStream<TaggedUnion<T1, T2>, KEY>(unionStream, unionKeySelector, keyType).window(windowAssigner);if (trigger != null) {windowedStream.trigger(trigger);}if (evictor != null) {windowedStream.evictor(evictor);}if (allowedLateness != null) {windowedStream.allowedLateness(allowedLateness);}//判断lateDataOutputTag是否为NULL，如果不为NULL，则调用windowedStream//的sideOutputLateData方法，传入lateDataOutputTag让迟到数据输出到测输出流中if (lateDataOutputTag != null) {windowedStream.sideOutputLateData(lateDataOutputTag);}return windowedStream.apply(new CoGroupWindowFunction<T1, T2, T, KEY, W>(function), resultType);}

3.5开放UnionTypeInfo类的public权限

该类就是union之后的公共类的类型 oneType代表Input1流的数据类型，TwoType代表Input2流的数据类型

3.6编译Flink源码flink-streaming-java模块

进入到flink-streaming-java所在磁盘目录输入以下命令编译

mvn clean install -DskipTests -Dfast

编译成功

3.7项目中查看maven是否已经刷新为最新代码

编译之后，可以看到导入的maven包已经有了新增的NewCoGroupedStreams类了，注意项目中的maven依赖中的flink版本，要与编译源码的版本一致，否则无法引入到。

4.测试

新建两个流，通过new NewCoGroupedStreams创建对象，在allowedLateness之后通过sideOutputLateData设置outputTag，然后通过with方法触发业务，with底层也是调用了apply，只不过他帮我们把返回的流转为了SingleOutputStreamOperator类型，可以用于提取测输出流。最后通过with.getSideOutput(outputTag)提取测输出流，最后通过map转换为 Tuple2<Integer, WaterSensor> 类型进行打印

    OutputTag<NewCoGroupedStreams.TaggedUnion<WaterSensor, WaterSensor>> outputTag = new OutputTag<>("later",new NewCoGroupedStreams.UnionTypeInfo<>(Types.POJO(WaterSensor.class), Types.POJO(WaterSensor.class)));NewCoGroupedStreams<WaterSensor, WaterSensor> newCgroupStream = new NewCoGroupedStreams<>(ds1, ds2);SingleOutputStreamOperator<String> with = newCgroupStream.where((x) -> x.getId()).equalTo(x -> x.getId()).window(TumblingEventTimeWindows.of(Time.seconds(10))).allowedLateness(Time.seconds(3)).sideOutputLateData(outputTag).with(new RichCoGroupFunction<WaterSensor, WaterSensor, String>() {@Overridepublic void coGroup(Iterable<WaterSensor> first, Iterable<WaterSensor> second, Collector<String> out) throws Exception {out.collect(first.toString() + "======" + second.toString());}});with.print();with.getSideOutput(outputTag).map(new MapFunction<NewCoGroupedStreams.TaggedUnion<WaterSensor, WaterSensor>, Tuple2<Integer, WaterSensor>>() {@Overridepublic Tuple2<Integer, WaterSensor> map(NewCoGroupedStreams.TaggedUnion<WaterSensor, WaterSensor> value) throws Exception {return value.isOne() ? Tuple2.of(1, value.getOne()) : Tuple2.of(2, value.getTwo());}}).print();

可以看到下图结果，ts代表时间戳，第一个打印是RichCoGroupFunction打印，代表关闭了1~10s的时间窗，后面我们在输入,WaterSensor{id='a', ts=1, vc=1} 就通过测输出流打印为二元组了

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

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

相关文章

SpringCloud中网关实现笔记

SpringCloud中网关实现笔记

SpringCloud中网关实现笔记 SpringCloudGateway&#xff1a;基于Spring的WebFlux技术&#xff0c;完全支持响应式编程&#xff0c;吞吐能力更强 Spring Cloud Gateway 是 Spring Cloud 的一个全新项目&#xff0c;用于构建基于 Spring 框架的 API 网关服务。它基于 Spring 5&a…
阅读更多...
Python爬虫-批量爬取星巴克全国门店

Python爬虫-批量爬取星巴克全国门店

前言 本文是该专栏的第22篇，后面会持续分享python爬虫干货知识，记得关注。 本文笔者以星巴克为例，通过Python实现批量爬取目标城市的门店数据以及全国的门店数据。 具体的详细思路以及代码实现逻辑，跟着笔者直接往下看正文详细内容。（附带完整代码） 正文 地址：aHR0cHM…
阅读更多...
基于SpringBoot和Vue的课程作业管理系统的设计与实现

基于SpringBoot和Vue的课程作业管理系统的设计与实现

今天要和大家聊的是一款基于SpringBoot和Vue的课程作业管理系统的设计与实现。 &#xff01;&#xff01;&#xff01; 有需要的小伙伴可以通过文章末尾名片咨询我哦&#xff01;&#xff01;&#xff01; &#x1f495;&#x1f495;作者&#xff1a;李同学 &#x1f495;&am…
阅读更多...
福昕阅读器 PDF 文档基本操作

福昕阅读器 PDF 文档基本操作

福昕阅读器 PDF 文档基本操作 References 转至 PDF 顶部 快捷键&#xff1a;Home. 转至 PDF 顶部 快捷键&#xff1a;End. 打开超链接 文本选择工具 -> 手形工具 (Hand Tool) -> 点击超链接 福昕阅读器 同时在多个窗口中打开多个文件 文件 -> 偏好设置 -> 文…
阅读更多...
开源博客项目Blog .NET Core源码学习（10：App.Framwork项目结构分析）

开源博客项目Blog .NET Core源码学习（10：App.Framwork项目结构分析）

开源博客项目Blog的解决方案总共包括4个项目&#xff0c;其中App.Hosting项目包括所有的页面及控制器类&#xff0c;其它项目主要提供数据库访问、基础类型定义等。这四个项目的依赖关系如下图所示&#xff0c;本文主要分析App.Framwork项目的主要结构及主要文件的用途。   …
阅读更多...
老胡的周刊（第134期）

老胡的周刊（第134期）

老胡的信息周刊[1]&#xff0c;记录这周我看到的有价值的信息&#xff0c;主要针对计算机领域&#xff0c;内容主题极大程度被我个人喜好主导。这个项目核心目的在于记录让自己有印象的信息做一个留存以及共享。 &#x1f3af; 项目 Llama-Chinese[2] 专注于 Llama 模型在中文方…
阅读更多...
nodejs+vue高校社团管理小程序的设计与实现python-flask-django-php

nodejs+vue高校社团管理小程序的设计与实现python-flask-django-php

相比于以前的传统手工管理方式&#xff0c;智能化的管理方式可以大幅降低学校的运营人员成本&#xff0c;实现了高校社团管理的标准化、制度化、程序化的管理&#xff0c;有效地防止了高校社团管理的随意管理&#xff0c;提高了信息的处理速度和精确度&#xff0c;能够及时、准…
阅读更多...
sentinel中StatisticSlot数据采集的原理

sentinel中StatisticSlot数据采集的原理

StatisticSlot数据采集的原理 时间窗口 固定窗口 在固定的时间窗口内&#xff0c;可以允许固定数量的请求进入&#xff1b;超过数量就拒绝或者排队&#xff0c;等下一个时间段进入, 如下图 时间窗长度划分为1秒 单个时间窗的请求阈值为3 上述存在一个问题, 假如9:18:04:…
阅读更多...
8个常见的数据可视化错误以及如何避免它们

8个常见的数据可视化错误以及如何避免它们

在当今以数据驱动为主导的世界里&#xff0c;清晰且具有洞察力的数据可视化至关重要。然而&#xff0c;在创建数据可视化时很容易犯错误&#xff0c;这可能导致对数据的错误解读。本文将探讨一些常见的糟糕数据可视化示例&#xff0c;并提供如何避免这些错误的建议。 本文总结了…
阅读更多...
前端学习之用css和html做一个仿淘宝的导航栏

前端学习之用css和html做一个仿淘宝的导航栏

代码 <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><title>仿淘宝界面案例</title><style>/* 最外层盒子 */.container{width: 270px;height: 385px;border: 1px solid rgb(255, 208, 0);bord…
阅读更多...
服务消费微服务

服务消费微服务

文章目录 1.示意图2.环境搭建1.创建会员消费微服务模块2.删除不必要的两个文件3.检查父子模块的pom.xml文件1.子模块2.父模块 4.pom.xml 添加依赖&#xff08;刷新&#xff09;5.application.yml 配置监听端口和服务名6.com/sun/springcloud/MemberConsumerApplication.java 创…
阅读更多...
LeetCode每日一题——移除链表元素

LeetCode每日一题——移除链表元素

移除链表元素OJ链接&#xff1a;203. 移除链表元素 - 力扣&#xff08;LeetCode&#xff09; 题目&#xff1a; 思路&#xff1a; 这与之前的移除元素的题目很相似&#xff0c;那么我们同样可以用类似的做法&#xff08;双指针&#xff09;进行解题。但是这是一个链表删除&a…
阅读更多...
代码随想录算法训练营Day56 ||leetCode 583. 两个字符串的删除操作 || 72. 编辑距离

代码随想录算法训练营Day56 ||leetCode 583. 两个字符串的删除操作 || 72. 编辑距离

647. 回文子串 dp[i][j]表示第i位开始&#xff0c;第j位结束的字符串是否为回文串 class Solution { public:int countSubstrings(string s) {vector<vector<bool>> dp(s.size(), vector<bool>(s.size(), false));int result 0;for (int i s.size() - 1…
阅读更多...
大白话扩散模型（无公式版）

大白话扩散模型（无公式版）

背景 传统的图像生成模型有GAN&#xff0c;VAE等&#xff0c;但是存在模式坍缩&#xff0c;即生成图片缺乏多样性&#xff0c;这是因为模型本身结构导致的。而扩散模型拥有训练稳定&#xff0c;保持图像多样性等特点&#xff0c;逐渐成为现在AIGC领域的主流。 扩散模型 正如…
阅读更多...
笔记本和台式机主板内部结构分析

笔记本和台式机主板内部结构分析

笔记本和态势机主板内存接口以及配件安装位置 笔记本主板 1 以thinkpad L-490为例,使用拆机小工具拆机&#xff0c;打开后面板&#xff0c;内部结构示意图如下 台式机主板 以技嘉-B660M-AORUS-PRO-AX型号主板为例 笔记本电脑和台式机电脑的相同之处 CPU&#xff1a;笔记本…
阅读更多...
牛客题霸-SQL篇（刷题记录三）

牛客题霸-SQL篇（刷题记录三）

本文基于前段时间学习总结的 MySQL 相关的查询语法&#xff0c;在牛客网找了相应的 MySQL 题目进行练习&#xff0c;以便加强对于 MySQL 查询语法的理解和应用。 由于涉及到的数据库表较多&#xff0c;因此本文不再展示&#xff0c;只提供 MySQL 代码与示例输出。 以下内容是…
阅读更多...
字母大小写转换

字母大小写转换

#include <stdio.h>//字母大小写转换 int main() {char ch 0;while(scanf("%c",&ch) 1){if(ch > a && ch < z)printf("%c\n",ch-32);if(ch > A && ch < Z)printf("%c\n",ch32);getchar();//处理\n}retu…
阅读更多...
GA遗传算法和ALNS算法的区别(我的APS项目七)

GA遗传算法和ALNS算法的区别(我的APS项目七)

博主用最简单的方式告诉你遗传算法是什么&#xff0c;估计这是网上最简单的遗传算法入门教程了。首先我们先带入一个问题&#xff0c;我们要去9大城市旅游&#xff0c;想知道每个城市走一遍&#xff0c;总路程最短的出行顺序是什么&#xff1f; OK&#xff0c;题目我们已经明确…
阅读更多...
STL标准模板库（C++

STL标准模板库（C++

在C里面有已经写好的标准模板库〈Standard Template Library)&#xff0c;就是我们常说的STL库&#xff0c;实现了集合、映射表、栈、队列等数据结构和排序、查找等算法。我们可以很方便地调用标准库来减少我们的代码量。 size/empty 所有的STL容器都支持这两个方法&#xff0c…
阅读更多...
力扣74---搜索二维矩阵

力扣74---搜索二维矩阵

目录 题目描述&#xff1a; 思路&#xff1a; 代码&#xff1a; 题目描述&#xff1a; 给你一个满足下述两条属性的 m x n 整数矩阵&#xff1a; 每行中的整数从左到右按非严格递增顺序排列。 每行的第一个整数大于前一行的最后一个整数。 给你一个整数 targ…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023