Flink之SideOutput(数据分流)
Flink在早期版本有一个split算子用来做数据分流使用的,但是在flink-1.12开始这个API就已经被删除了,在1.12版本以后我们是通过process算子来做数据分流的,这里就介绍一下如何使用prodess进行数据分流.
- 代码
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;/*** @Author: J* @Version: 1.0* @CreateTime: 2023/8/7* @Description: 测流输出**/
public class FlinkSideOutput {public static void main(String[] args) throws Exception {// 构建流环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 设置并行度env.setParallelism(3);// 这里使用的是自定义数据源为了方便测试,具体数据源根据自己的实际情况进行更换DataStreamSource<CustomizeBean> customizeSourceStream = env.addSource(new CustomizeSource());/*** 需求* 1. 将性别为M且爱好为'羽毛球运动爱好者'分到一个流* 2. 将性别为W且爱好为'篮球运动爱好者'或'钓鱼爱好者'分到一个流* 3. 其他保留到主流**/SingleOutputStreamOperator<CustomizeBean> processedStream = customizeSourceStream.process(new ProcessFunction<CustomizeBean, CustomizeBean>() {@Overridepublic void processElement(CustomizeBean value, ProcessFunction<CustomizeBean, CustomizeBean>.Context ctx, Collector<CustomizeBean> out) throws Exception {String gender = value.getGender(); // 性别String hobbit = value.getHobbit(); // 爱好if (gender.equals("M") && hobbit.equals("羽毛球运动爱好者")) {// 将性别为M且爱好为'羽毛球运动爱好者'进行分流, 注意这里要声明类型,Java无法自行推断ctx.output(new OutputTag<CustomizeBean>("M-羽毛球", TypeInformation.of(CustomizeBean.class)), value);} else if (gender.equals("W") && (hobbit.equals("篮球运动爱好者") || hobbit.equals("钓鱼爱好者"))) {// 将性别为W且爱好为'篮球运动爱好者'或'钓鱼爱好者'进行分流, 注意这里要声明类型,Java无法自行推断ctx.output(new OutputTag<CustomizeBean>("W-篮球/钓鱼", TypeInformation.of(CustomizeBean.class)), value);} else {// 将剩下的数据保留在主流中out.collect(value);}}});// 获取'M-羽毛球'分流数据,这里也要加上类型声明DataStream<CustomizeBean> mSideOutput = processedStream.getSideOutput(new OutputTag<CustomizeBean>("M-羽毛球", TypeInformation.of(CustomizeBean.class)));// 打印'M-羽毛球'结果mSideOutput.print("M-羽毛球");// 获取'W-篮球/钓鱼'分流数据,这里也要加上类型声明DataStream<CustomizeBean> wSideOutput = processedStream.getSideOutput(new OutputTag<CustomizeBean>("W-篮球/钓鱼", TypeInformation.of(CustomizeBean.class)));// 打印结果wSideOutput.print("W-篮球/钓鱼");// 主流数据打印结果processedStream.print("主数据流");env.execute("Side Output");}
}
- 结果数据
主数据流:2> CustomizeBean(name=AAA-641, age=44, gender=W, hobbit=非遗文化爱好者)
主数据流:3> CustomizeBean(name=AAA-17, age=62, gender=M, hobbit=书法爱好者)
主数据流:1> CustomizeBean(name=AAA-429, age=25, gender=W, hobbit=非遗文化爱好者)
主数据流:2> CustomizeBean(name=AAA-218, age=33, gender=M, hobbit=旅游爱好者)
主数据流:3> CustomizeBean(name=AAA-826, age=39, gender=M, hobbit=篮球运动爱好者)
主数据流:1> CustomizeBean(name=AAA-190, age=31, gender=M, hobbit=旅游爱好者)
主数据流:2> CustomizeBean(name=AAA-266, age=32, gender=W, hobbit=网吧战神)
主数据流:3> CustomizeBean(name=AAA-106, age=70, gender=M, hobbit=书法爱好者)
主数据流:1> CustomizeBean(name=AAA-911, age=50, gender=M, hobbit=网吧战神)
M-羽毛球:2> CustomizeBean(name=AAA-925, age=65, gender=M, hobbit=羽毛球运动爱好者)
主数据流:3> CustomizeBean(name=AAA-20, age=59, gender=M, hobbit=书法爱好者)
主数据流:1> CustomizeBean(name=AAA-409, age=79, gender=W, hobbit=天文知识爱好者)
主数据流:2> CustomizeBean(name=AAA-865, age=58, gender=W, hobbit=天文知识爱好者)
主数据流:3> CustomizeBean(name=AAA-898, age=33, gender=M, hobbit=天文知识爱好者)
主数据流:1> CustomizeBean(name=AAA-85, age=38, gender=W, hobbit=非遗文化爱好者)
主数据流:2> CustomizeBean(name=AAA-883, age=51, gender=M, hobbit=美食爱好者)
主数据流:3> CustomizeBean(name=AAA-243, age=37, gender=M, hobbit=钓鱼爱好者)
主数据流:1> CustomizeBean(name=AAA-430, age=28, gender=W, hobbit=旅游爱好者)
主数据流:2> CustomizeBean(name=AAA-127, age=65, gender=W, hobbit=网吧战神)
W-篮球/钓鱼:3> CustomizeBean(name=AAA-986, age=52, gender=W, hobbit=钓鱼爱好者)
主数据流:1> CustomizeBean(name=AAA-840, age=50, gender=W, hobbit=旅游爱好者)
M-羽毛球:2> CustomizeBean(name=AAA-196, age=34, gender=M, hobbit=羽毛球运动爱好者)
主数据流:3> CustomizeBean(name=AAA-142, age=46, gender=W, hobbit=乒乓球运动爱好者)
主数据流:1> CustomizeBean(name=AAA-985, age=78, gender=W, hobbit=美食爱好者)
W-篮球/钓鱼:2> CustomizeBean(name=AAA-490, age=50, gender=W, hobbit=钓鱼爱好者)
主数据流:3> CustomizeBean(name=AAA-295, age=77, gender=M, hobbit=篮球运动爱好者)
主数据流:1> CustomizeBean(name=AAA-754, age=50, gender=M, hobbit=天文知识爱好者)
主数据流:2> CustomizeBean(name=AAA-249, age=35, gender=W, hobbit=羽毛球运动爱好者)
W-篮球/钓鱼:3> CustomizeBean(name=AAA-908, age=27, gender=W, hobbit=钓鱼爱好者)
主数据流:1> CustomizeBean(name=AAA-674, age=73, gender=M, hobbit=非遗文化爱好者)
通过结果内容可以看到数据完全按照我们分流的逻辑进行输出的,如果想在主数据流中讲所有数据保留下来,Collector<Object> out单独拎出来即可,也就是不加到判断逻辑中,代码如下,这里就只展示部分代码了
SingleOutputStreamOperator<CustomizeBean> processedStream = customizeSourceStream.process(new ProcessFunction<CustomizeBean, CustomizeBean>() {@Overridepublic void processElement(CustomizeBean value, ProcessFunction<CustomizeBean, CustomizeBean>.Context ctx, Collector<CustomizeBean> out) throws Exception {String gender = value.getGender(); // 性别String hobbit = value.getHobbit(); // 爱好// 将所有数据保留在主流中out.collect(value);// 开始进行分流处理if (gender.equals("M") && hobbit.equals("羽毛球运动爱好者")) {// 将性别为M且爱好为'羽毛球运动爱好者'进行分流, 注意这里要声明类型,Java无法自行推断ctx.output(new OutputTag<CustomizeBean>("M-羽毛球", TypeInformation.of(CustomizeBean.class)), value);} else if ((gender.equals("W") && (hobbit.equals("篮球运动爱好者")) || (gender.equals("W") && hobbit.equals("钓鱼爱好者")))) {// 将性别为W且爱好为'篮球运动爱好者'或'钓鱼爱好者'进行分流, 注意这里要声明类型,Java无法自行推断ctx.output(new OutputTag<CustomizeBean>("W-篮球/钓鱼", TypeInformation.of(CustomizeBean.class)), value);}}});
所有的内容到这里就结束了.
相关文章:
Flink之SideOutput(数据分流)
Flink在早期版本有一个split算子用来做数据分流使用的,但是在flink-1.12开始这个API就已经被删除了,在1.12版本以后我们是通过process算子来做数据分流的,这里就介绍一下如何使用prodess进行数据分流. 代码 import org.apache.flink.api.common.typeinfo.TypeInformation; im…...
Android Studio新版本logcat过滤说明
按包名过滤 //输入package:(输入一个p就会有提示的) ,后面加上包名 比如: package:com.xal.runcontrol package:包名可以完整或者输部分包名即可 package:包名需要输完整准确 package~:正则表达式过滤 不了解正则表达式的可以参考&#…...
carsim与matlab仿真
matlab2021a安装教程,亲测。 百度网盘: matlab2021a安装包 提取码:1223 CarSim2020安装教程, 亲测。 百度网盘: CarSim2020安装包 提取码:1223 ,破解可参考 b站视频...
rust里如何快速实现一个LRU 本地缓存?
LRU是Least Recently Used(最近最少使用)的缩写,是一种常见的缓存淘汰算法。LRU算法的基本思想是,当缓存空间已满时,优先淘汰最近最少使用的数据,以保留最常用的数据。 在计算机系统中,LRU算法…...
MQTT 订阅接收消息 mosquitto 方式
1 说明 采用 mosquitto 库,实现订阅主题,并接收消息。其中服务器有做限制,需要对应的 cilent id ,cafile 、certfile 、keyfile 等配置2 环境 采用ubuntu 直接编译调试 安装mosquitto 库 sudo apt install libmosquitto-dev su…...
以mod_jk方式整合apache与tomcat(动静分离)
前言: 为什么要整合apache和tomcat apache对静态页面的处理能力强,而tomcat对静态页面的处理不如apache,整合后有以下好处 提升对静态文件的处理性能 利用 Web 服务器来做负载均衡以及容错 更完善地去升级应用程序 jk整合方式介绍&#…...
springboot动态数据源切换
1)、就是将多个数据源全部注入到bean中,根据需要实现多数据源之间的切换。 2)、使用baomidou的DS注解。见文章DS注解实现数据源动态切换 com.baomidou dynamic-datasource-spring-boot-starter 3.5.1 ##设置默认的数据源或者数据源组,默认值…...
代码随想录训练营day14
101. 对称二叉树 给你一个二叉树的根节点 root , 检查它是否轴对称。 func isSymmetric(root *TreeNode) bool {if root nil{ return true}return judge(root.Left,root.Right) }func judge(lf *TreeNode , ri *TreeNode)bool{if lf nil && ri nil{ retu…...
功能测试进阶自动化测试如何摸清学习方向,少走弯路呢?
目录 抛开疑问,只做学术探讨 小白在想什么? 盖楼之前先打好地基,首先需要学习一门语言 语言入门后,正式踏上开始自动化成神之路,入门篇Selenium 玩腻了Selenium 开始接触自动化框架unittest/testNG 不满足于单元…...
检测前端是否可以ping通后端返回的ip地址
检测前端是否可以ping通后端返回的ip地址 前端检测是否可ping通ip地址(PC端)前端检测是否可ping通ip地址(uniapp小程序端) 前端检测是否可ping通ip地址(PC端) // 前端检测是否可ping通ip地址 ping…...
SMART司马他法则(目标管理)
S代表具体(Specific),指绩效考核要切中特定的工作指标,不能笼统; M代表可度量(Measurable),指绩效指标是数量化或者行为化的,验证这些绩效指标的数据或者信息是可以获得的; A代表可实现(Attainable)&…...
【LeetCode】删除并获得点数
删除并获得点数 题目描述算法分析编程代码空间优化 链接: 删除并获得点数 题目描述 算法分析 编程代码 class Solution { public:int deleteAndEarn(vector<int>& nums) {const int N 10001;int arr[N] {0};for(const auto& n : nums){arr[n]n;}vector<in…...
SciencePub学术 | 传感器类重点SCIE征稿中
SciencePub学术 刊源推荐: 传感器类重点SCIE征稿中!信息如下,录满为止: 一、期刊概况: 传感器类重点SCIE 【期刊简介】IF:2.0-2.5,JCR3区,中科院4区; 【版面类型】正刊࿱…...
移动端开发基础总结
移动端学习总结 (适合于复习) 移动端基础 技术选型: 单独制作移动端页面(主流) 流式布局(百分比布局)flex弹性布局(强烈推荐)lessrem媒体查询布局混合布局 响应式页面兼容移动端(…...
小X学游泳(深搜)
第一题 题目描述 小X想要学游泳。 这天,小X来到了游泳池,发现游泳池可以用N行M列的格子来表示,每个格子的面积都是1,且格子内水深相同。 由于小X刚刚入门,他只能在水深相同的地方游泳。为此,他把整个游泳池…...
分布式协议与算法——拜占庭将军问题
拜占庭将军问题 背景:以战国时期为背景 战国时期,齐、楚、燕、韩、赵、魏、秦七雄并立,后来秦国的势力不断强大起来,成了东方六国的共同威胁。于是,这六个国家决定联合,全力抗秦,免得被秦国各个…...
MySQL数据库管理的基本原则和技巧
MySQL数据库是一种常用的关系型数据库管理系统,用于存储和管理大量的数据。在进行MySQL数据库管理时,有一些基本原则和技巧可以帮助我们更有效地管理数据库。 数据库设计原则: 合理规划数据表结构: 根据数据之间的关系和业务需求…...
SQL-每日一题【1193. 每月交易 I】
题目 Table: Transactions 编写一个 sql 查询来查找每个月和每个国家/地区的事务数及其总金额、已批准的事务数及其总金额。 以 任意顺序 返回结果表。 查询结果格式如下所示。 示例 1: 解题思路 1.题目要求我们查找每个月和每个国家/地区的事务数及其总金额、已批准的事务数…...
探析青少年口才训练在个人发展中的重要性与影响
论文题目:探析青少年口才训练在个人发展中的重要性与影响 摘要: 本论文旨在探讨青少年口才训练对个人发展的重要性和影响。通过对相关文献的综述和实证研究的分析,论文将阐述口才训练对青少年自信心、表达能力和思维能力的提升,以…...
HTML 元素的 class 和 id 属性有何区别?
聚沙成塔每天进步一点点 ⭐ 专栏简介⭐ 唯一性⭐ 选择器权重⭐ JS操作⭐ CSS和JavaScript引用⭐ 写在最后 ⭐ 专栏简介 前端入门之旅:探索Web开发的奇妙世界 记得点击上方或者右侧链接订阅本专栏哦 几何带你启航前端之旅 欢迎来到前端入门之旅!这个专栏…...
Docker 离线安装指南
参考文章 1、确认操作系统类型及内核版本 Docker依赖于Linux内核的一些特性,不同版本的Docker对内核版本有不同要求。例如,Docker 17.06及之后的版本通常需要Linux内核3.10及以上版本,Docker17.09及更高版本对应Linux内核4.9.x及更高版本。…...
线程同步:确保多线程程序的安全与高效!
全文目录: 开篇语前序前言第一部分:线程同步的概念与问题1.1 线程同步的概念1.2 线程同步的问题1.3 线程同步的解决方案 第二部分:synchronized关键字的使用2.1 使用 synchronized修饰方法2.2 使用 synchronized修饰代码块 第三部分ÿ…...
页面渲染流程与性能优化
页面渲染流程与性能优化详解(完整版) 一、现代浏览器渲染流程(详细说明) 1. 构建DOM树 浏览器接收到HTML文档后,会逐步解析并构建DOM(Document Object Model)树。具体过程如下: (…...
JUC笔记(上)-复习 涉及死锁 volatile synchronized CAS 原子操作
一、上下文切换 即使单核CPU也可以进行多线程执行代码,CPU会给每个线程分配CPU时间片来实现这个机制。时间片非常短,所以CPU会不断地切换线程执行,从而让我们感觉多个线程是同时执行的。时间片一般是十几毫秒(ms)。通过时间片分配算法执行。…...
【JavaSE】绘图与事件入门学习笔记
-Java绘图坐标体系 坐标体系-介绍 坐标原点位于左上角,以像素为单位。 在Java坐标系中,第一个是x坐标,表示当前位置为水平方向,距离坐标原点x个像素;第二个是y坐标,表示当前位置为垂直方向,距离坐标原点y个像素。 坐标体系-像素 …...
OpenLayers 分屏对比(地图联动)
注:当前使用的是 ol 5.3.0 版本,天地图使用的key请到天地图官网申请,并替换为自己的key 地图分屏对比在WebGIS开发中是很常见的功能,和卷帘图层不一样的是,分屏对比是在各个地图中添加相同或者不同的图层进行对比查看。…...
MySQL账号权限管理指南:安全创建账户与精细授权技巧
在MySQL数据库管理中,合理创建用户账号并分配精确权限是保障数据安全的核心环节。直接使用root账号进行所有操作不仅危险且难以审计操作行为。今天我们来全面解析MySQL账号创建与权限分配的专业方法。 一、为何需要创建独立账号? 最小权限原则…...
安宝特方案丨船舶智造的“AR+AI+作业标准化管理解决方案”(装配)
船舶制造装配管理现状:装配工作依赖人工经验,装配工人凭借长期实践积累的操作技巧完成零部件组装。企业通常制定了装配作业指导书,但在实际执行中,工人对指导书的理解和遵循程度参差不齐。 船舶装配过程中的挑战与需求 挑战 (1…...
音视频——I2S 协议详解
I2S 协议详解 I2S (Inter-IC Sound) 协议是一种串行总线协议,专门用于在数字音频设备之间传输数字音频数据。它由飞利浦(Philips)公司开发,以其简单、高效和广泛的兼容性而闻名。 1. 信号线 I2S 协议通常使用三根或四根信号线&a…...
招商蛇口 | 执笔CID,启幕低密生活新境
作为中国城市生长的力量,招商蛇口以“美好生活承载者”为使命,深耕全球111座城市,以央企担当匠造时代理想人居。从深圳湾的开拓基因到西安高新CID的战略落子,招商蛇口始终与城市发展同频共振,以建筑诠释对土地与生活的…...
