当前位置: 首页 > news >正文

Flink流计算处理-旁路输出

使用Flink做流数据处理时,除了主流数据输出,还自定义侧流输出即旁路输出,以实现灵活的数据拆分。

定义旁路输出标签

首先需要定义一个OutputTag,代码如下:

// 这需要是一个匿名的内部类,以便我们分析类型
OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

在ProcessFunction使用Context调用

可以通过以下Function中,将outputTag作为参数传递到Context中

  • ProcessFunction
  • KeyedProcessFunction
  • CoProcessFunction
  • KeyedCoProcessFunction
  • ProcessWindowFunction
  • ProcessAllWindowFunction

代码示例:

DataStream<Integer> input = ...;final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};SingleOutputStreamOperator<Integer> mainDataStream = input.process(new ProcessFunction<Integer, Integer>() {@Overridepublic void processElement(Integer value,Context ctx,Collector<Integer> out) throws Exception {// 发送数据到主要的输出out.collect(value);// 发送数据到旁路输出ctx.output(outputTag, "sideout-" + String.valueOf(value));}});

在 DataStream 运算结果上使用 getSideOutput(OutputTag) 方法获取旁路输出流:

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};
SingleOutputStreamOperator<Integer> mainDataStream = ...;
// 获取到侧流输出DataStream,输出结果类型要与outputTag 定义的一致
DataStream<String> sideOutputStream = mainDataStream.getSideOutput(outputTag);

本文中只列出了Java代码的实现;
Flink官网还有Scala/python代码实现

参考链接:https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/dev/datastream/side_output/

相关文章:

Flink流计算处理-旁路输出

使用Flink做流数据处理时&#xff0c;除了主流数据输出&#xff0c;还自定义侧流输出即旁路输出&#xff0c;以实现灵活的数据拆分。 定义旁路输出标签 首先需要定义一个OutputTag&#xff0c;代码如下&#xff1a; // 这需要是一个匿名的内部类&#xff0c;以便我们分析类型…...

nginx正向代理的配置和使用

nginx正向代理的配置和使用 nginx正向代理的配置和使用nginx正向代理的配置和使用安装包准备下载nginx安装包下载正向代理模块的包版本与模块对照表部署nginx服务上传nginx包和正向模块包解压,改名安装nginx配置正向代理创建nginx用户检查nginx配置并启动nginx服务所在服务器验…...

Oracle Trace File Analyzer 介绍及简单使用

一、什么是Oracle Trace File Analyzer Oracle Autonomous Health Framework(AHF) 包含 Oracle ORAchk, Oracle EXAchk, and Oracle Trace File Analyzer(TFA). AHF工具包包含了Oracle常用的多种诊断工具&#xff0c;如 ORAchk, Oracle EXAchk, and Oracle Trace File Analyzer…...

面试实战篇 | 快手本地生活,结合项目谈Redis实战项目场景?MySQL InnoDB存储引擎如何工作的?策略模式?

本期是【你好&#xff0c;面试官】系列文章的第21期&#xff0c;持续更新中…。 《你好&#xff0c;面试官》系列目前已经连载20篇了&#xff0c;据说看了这个系列的朋友都拿到了大厂offer~ 你好&#xff0c;面试官 | 你真的理解面向 “对象”&#xff1f;你好&#xff0c;面…...

Hadoop之——WordCount案例与执行本地jar包

目录 一、WordCount代码 (一)WordCount简介 1.wordcount.txt (二)WordCount的java代码 1.WordCountMapper 2.WordCountReduce 3.WordCountDriver (三)IDEA运行结果 (四)Hadoop运行wordcount 1.在HDFS上新建一个文件目录 2.新建一个文件&#xff0c;并上传至该目录下…...

利用git reflog 命令来查看历史提交记录,并使用提交记录恢复已经被删除掉的分支

一.问题描述 当我们在操作中手误删除了某个分支&#xff0c;那该分支中提交的内容也没有了&#xff0c;我们可以利用git reflog这个命令来查看历史提交的记录从而恢复被删除的分支和提交的内容 二.模拟问题 1.创建git仓库&#xff0c;并提交一个文件 [rootcentos7-temp /da…...

【软件测试】大厂测试开发你真的了解吗?测试开发养成记......

目录&#xff1a;导读前言一、Python编程入门到精通二、接口自动化项目实战三、Web自动化项目实战四、App自动化项目实战五、一线大厂简历六、测试开发DevOps体系七、常用自动化测试工具八、JMeter性能测试九、总结&#xff08;尾部小惊喜&#xff09;前言 在一些大公司里&…...

Redis中的hash结构和扩容机制

1.rehash原理 hash包含两个数据结构为字典数组ht[0]和ht[1]。其中ht[0]用来存放数据&#xff0c;ht[1]在rehash时使用。 扩容时&#xff0c;ht[1]的大小为第一个大于等于ht[0].used*2的2的幂次方的数&#xff1b; 收缩时&#xff0c;ht[1]的大小为第一个大于等于ht[0].used的…...

【C++奇技淫巧】前置自增与后置自增的区别(++i,i++)【2023.02.08】

简介 先说i和i的区别&#xff0c;判断语句中if(i)是拿i的值先判断&#xff0c;而后自增&#xff1b;if(i)是先自增i再进行判断。涉及到左值与右值也有点区别&#xff0c;i返回的是右值&#xff0c;i返回的是左值。也就是下面的代码要解释的东西。 #include <iostream>i…...

实战打靶集锦-005-HL

**写在前面&#xff1a;**记录一次曲折的打靶经历。 目录1. 主机发现2. 端口扫描3. 服务枚举4. 服务探查4.1 浏览器访问4.2 目录枚举4.3 探查admin4.4 探查index4.5 探查login5 公共EXP搜索6. 再次目录枚举6.1 探查superadmin.php6.2 查看页面源代码6.3 base64绕过6.4 构建反弹…...

铁路系统各专业介绍(车机工电辆)

目录 1 车务段 1.1 职能简介 1.2 路段名单 1.3 岗位级别 2 机务段 2.1 职能简介 2.2 路段名单 2.3 岗位级别 3 工务段 3.1 职能简介 3.2 路段名单 3.3 岗位级别 4 电务段 4.1 职能简介 4.2 路段名单 4.3 岗位级别 5 车辆段 5.1 职能简介 5.2 路段名单 5.3 …...

2/11考试总结

时间安排 7:30–7:50 读题&#xff0c;T1貌似是个 dp &#xff0c;T2 数据结构&#xff0c;T3 可能是数据结构。 7:50–9:45 T1&#xff0c;点规模非常大&#xff0c;可以达到 1e18 级别&#xff0c;感觉应该没法直接做&#xff0c;考虑每条新增的边的贡献&#xff0c;想到用 …...

Java Set集合

7 Set集合 7.1 Set集合的概述和特点 Set集合的特点 不包含重复元素的集合没有带索引的方法&#xff0c;所以不能使用普通for循环 Set集合是接口通过实现类实例化&#xff08;多态的形式&#xff09; HashSet&#xff1a;添加的元素是无序&#xff0c;不重复&#xff0c;无索引…...

【手写 Vuex 源码】第七篇 - Vuex 的模块安装

一&#xff0c;前言 上一篇&#xff0c;主要介绍了 Vuex 模块收集的实现&#xff0c;主要涉及以下几个点&#xff1a; Vuex 模块的概念&#xff1b;Vuex 模块和命名空间的使用&#xff1b;Vuex 模块收集的实现-构建“模块树”&#xff1b; 本篇&#xff0c;继续介绍 Vuex 模…...

EOC第六章《块与中枢派发》

文章目录第37条&#xff1a;理解block这一概念第38条&#xff1a;为常用的块类型创建typedef第39条&#xff1a;用handler块降低代码分散程度第41条&#xff1a;多用派发队列&#xff0c;少用同步锁方案一&#xff1a;使用串行同步队列来将读写操作都安排到同一个队列里&#x…...

八、Git远程仓库操作——跨团队成员的协作

前言 前面一篇博文介绍了git团队成员之间的协作&#xff0c;现在在介绍下如果是跨团队成员的话&#xff0c;如何协作&#xff1f; 跨团队成员协作&#xff0c;其实就是你不属于那个项目的成员&#xff0c;你没有权限向那个仓库提交代码。但是github还有另一种 pull request&a…...

算法刷题打卡第88天:字母板上的路径

字母板上的路径 难度&#xff1a;中等 我们从一块字母板上的位置 (0, 0) 出发&#xff0c;该坐标对应的字符为 board[0][0]。 在本题里&#xff0c;字母板为board ["abcde", "fghij", "klmno", "pqrst", "uvwxy", "…...

UVa The Morning after Halloween 万圣节后的早晨 双向BFS

题目链接&#xff1a;The Morning after Halloween 题目描述&#xff1a; 给定一个二维矩阵&#xff0c;图中有障碍物和字母&#xff0c;你需要把小写字母移动到对应的大写字母位置&#xff0c;不同的小写字母可以同时移动&#xff08;上下左右四个方向或者保持不动 &#xff0…...

Connext DDS属性配置参考大全(3)

Transport传输dds.participant.logging.time_based_logging.process_received_messagedds.participant.logging.time_based_logging.process_received_message.timeout...

Docker-安装Jenkins-使用jenkins发版Java项目

文章目录0.前言环境背景1.操作流程1.1前期准备工作1.1.1环境变量的配置1.2使用流水线的方式进行发版1.2.1新建流水线任务1.2.2流水线操作工具tools步骤stages步骤1:拉取代码编译步骤2:发送文件并启动0.前言 学海无涯&#xff0c;旅“途”漫漫&#xff0c;“途”中小记&#xff…...

spring 中的 Bean 是否线程安全

文章目录结论1、spring中的Bean从哪里来&#xff1f;2、spring中什么样的Bean存在线程安全问题&#xff1f;3、如何处理spring Bean的线程安全问题&#xff1f;结论 其实&#xff0c;Spring 中的 Bean 是否线程安全&#xff0c;其实跟 Spring 容器本身无关。Spring框架中没有提…...

微电网两阶段鲁棒优化经济调度方法[3]【升级优化版本】(Matlab代码实现)

&#x1f4a5;&#x1f4a5;&#x1f4a5;&#x1f49e;&#x1f49e;&#x1f49e;欢迎来到本博客❤️❤️❤️&#x1f4a5;&#x1f4a5;&#x1f4a5; &#x1f3c6;博主优势&#xff1a;&#x1f31e;&#x1f31e;&#x1f31e;博客内容尽量做到思维缜密&#xff0c;逻辑…...

C++入门教程||C++ 数据类型||C++ 变量类型

C 数据类型 使用编程语言进行编程时&#xff0c;需要用到各种变量来存储各种信息。变量保留的是它所存储的值的内存位置。这意味着&#xff0c;当您创建一个变量时&#xff0c;就会在内存中保留一些空间。 您可能需要存储各种数据类型&#xff08;比如字符型、宽字符型、整型…...

【visio使用技巧】图片导出pdf时去掉多余空白

问题 在visio导出pdf格式的图片时&#xff0c;往往会存在多余的白边&#xff0c;如下图所示&#xff1a; 解决方法 依次点击&#xff1a;菜单栏→文件→选项→自定义功能区→勾选“开发工具”→确定。 依次点击菜单栏→开发工具→显示ShapeSheet→页→Print Properties→将…...

Rust语言之Option枚举类型

概述 Option是Rust语言设计中最重要的枚举类型之一&#xff0c;它编码了其它语言中空值与非空值的概念&#xff0c;差异在于&#xff0c;Rust不会允许你像其它语言一样以非空值的方式来使用一个空值&#xff0c;这避免了很多错误。Option在标准库中的定义如下&#xff1a; pu…...

基于TimeQuest时序优化原理和方法

&#x1f4a1; 回顾基于RTL逻辑时序优化的基本思路&#xff0c;在关键路径中插入寄存器来优化时序 分析最坏路径 通过前面对TimeQuest软件的理解&#xff0c;基本上可以找到关键路径&#xff0c;此文章主要对关键路径时序进行优化&#xff0c;使设计达到时序要求&#xff0c;以…...

LeetCode第332场周赛

2023.2.12LeetCode第332场周赛 6354. 找出数组的串联值 思路 双指针模拟&#xff0c;两个指针相遇的时候要特判 算法 class Solution { public:long long findTheArrayConcVal(vector<int>& nums) {long long ans 0;int i 0, j nums.size() - 1;while (i <…...

2023-2-12刷题情况

字母板上的路径 题目描述 我们从一块字母板上的位置 (0, 0) 出发&#xff0c;该坐标对应的字符为 board[0][0]。 在本题里&#xff0c;字母板为board [“abcde”, “fghij”, “klmno”, “pqrst”, “uvwxy”, “z”]&#xff0c;如下所示。 我们可以按下面的指令规则行动…...

拉普拉斯矩阵

拉普拉斯算子 Δff(xi1,yj)f(xi−1,yj)f(xi,yj1)f(xi,yj−1)−4f(xi,yj)∑(k,l)∈N(i,j)(f(xk,yl)−f(xi,yj))\begin{aligned} \Delta f & f\left(x_{i1}, y_j\right) f\left(x_{i-1},y_j\right) f\left(x_i,y_{j1}\right)f\left(x_i,y_{j-1}\right) - 4f\left(x_i,y_j\r…...

Top-1错误率、Top-5错误率等常见的模型算法评估指标解析

Top-1 错误率&#xff1a;指预测输出的概率最高的类别与人工标注的类别相符的准确率&#xff0c;就是你预测的label取最后概率向量里面最大的那一个作为预测结果&#xff0c;如过你的预测结果中概率最大的那个分类正确&#xff0c;则预测正确&#xff0c;否则预测错误。比如预测…...

合肥城乡建设委员会网站/安徽网络seo

《MYSQL数据库mysql连接数设置操作方法(Too many connections)》要点&#xff1a;本文介绍了MYSQL数据库mysql连接数设置操作方法(Too many connections)&#xff0c;希望对您有用。如果有疑问&#xff0c;可以联系我们。mysql在使用过程中,发现连接数超了~~~~MYSQL必读[rootli…...

宝塔一键部署wordpress最新版/sem优化和seo的区别

文章目录&#xff08;一&#xff09;用户自定义的环境变量1、环境变量设置2、查询自定义环境变量&#xff08;1&#xff09;env命令&#xff08;2&#xff09;set命令3、删除自定义环境变量&#xff08;二&#xff09;系统自带的环境变量——PATH环境变量1、PATH环境变量2、PAT…...

b2b电子商务网站的盈利模式/seo职业规划

laizi:http://345947630-qq-com.iteye.com/blog/1293394 getServletConfig() 在servlet初始化时&#xff0c;容器传递进来一个ServletConfig对象并保存在servlet实例中&#xff0c;该对象允许访问两项内容&#xff1a;初始化参数和ServletContext对象&#xff0c;前者通常由容…...

中学生设计的网站/注册网站流程

1、借助于iframe首先&#xff0c;最容易想到的应该是使用iframe,虽然html5废除了frame,但是依旧保留了iframe,我们仍可以继续使用&#xff0c;iframe有一个frameboder属性&#xff0c;设置属性值为0或者为no,去除iframe的边框。然后将scrolling设为no。这是完全可行的&#xff…...

wordpress自制主题/微博推广方案

过去几年&#xff0c;关于文本编辑框(Edit)控件的响应事件&#xff0c;我都是在主程序 while(GetMessage(&messages, NULL, 0, 0)) { ... } 捕获。 总感觉这种方式让人感觉不舒服。 今天想给一个 Edit 控件加一下按下【回车】的响应事件&#xff0c;我百度了一下&#xff0…...

沈阳做网站的设计公司哪家好/百度seo排名优化联系方式

创造实验数据 // 现在的时间 Date nowDate new Date(); // 设置三天后的时间 Calendar c Calendar.getInstance(); c.setTime(nowDate); c.add(Calendar.DATE,3); Date afterThreeDay c.getTime(); 默认向下取整天数 // 一天的毫秒数 long DAY 24L * 60L * 60L * 1000L; …...