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

Flink多流处理之connect拼接流

Flink中的拼接流connect的使用其实非常简单,就是leftStream.connect(rightStream)的方式,但是有一点我们需要清楚,使用connect后并不是将两个流给串联起来了,而是将左流和右流建立一个联系,作为一个大的流,并且这个大的流可以使用相同的逻辑处理leftStreamrightStream,也可以使用不同的逻辑处理leftStreamrightStream.
如下图:
在这里插入图片描述

下面的演示代码也可以通过这个图结合来看,其实connect算子最主要的作用就是共享状态,如常用的广播状态.

  • 代码
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
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.co.CoMapFunction;import java.util.Arrays;/*** @Author: J* @Version: 1.0* @CreateTime: 2023/8/7* @Description: 多流操作-流连接**/
public class FlinkConnect {public static void main(String[] args) throws Exception {// 构建流环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 设置并行度env.setParallelism(3);// 添加数据源1DataStreamSource<String> sourceStream1 = env.fromCollection(Arrays.asList("a", "b", "c", "d"));// 添加数据源2DataStreamSource<Double> sourceStream2 = env.fromCollection(Arrays.asList(22.2, 11.0, 6.0, 98.0, 100.0));// 拼接数据流ConnectedStreams<String, Double> connectedStream = sourceStream1.connect(sourceStream2);// 这里使用map算子作为演示SingleOutputStreamOperator<String> resultStream = connectedStream.map(new CoMapFunction<String, Double, String>() {/*** map1作为左流**/@Overridepublic String map1(String value) throws Exception {return "字符串: " + value;}/*** map2作为右流**/@Overridepublic String map2(Double value) throws Exception {return "数字: " + (value * 100);}});// 打印结果resultStream.print();env.execute("Connect Operator");}
}
  • 结果
3> 字符串: b
1> 数字: 600.0
2> 字符串: a
3> 数字: 1100.0
2> 数字: 2220.0
2> 字符串: d
2> 数字: 9800.0
3> 数字: 10000.0
1> 字符串: c

相关文章:

Flink多流处理之connect拼接流

Flink中的拼接流connect的使用其实非常简单,就是leftStream.connect(rightStream)的方式,但是有一点我们需要清楚,使用connect后并不是将两个流给串联起来了,而是将左流和右流建立一个联系,作为一个大的流,并且这个大的流可以使用相同的逻辑处理leftStream和rightStream,也可以…...

对任意类型数都可以排序的函数:qsort函数

之前我们学习过冒泡排序&#xff1a; int main() {int arr[] { 9,7,8,6,5,4,3,2,1,0 };int sz sizeof(arr)/sizeof(arr[0]);int i 0;for (i 0; i < sz-1; i) {int j 0;for (j 0; j < sz-1-i; j) {if (arr[j] > arr[j 1]){int temp 0;temp arr[j];arr[j] ar…...

vue数据更新table内容不更新解决方法

场景&#xff1a; table组件绑定的数据变化时&#xff0c;页面没有重新渲染&#xff0c;常见于子组件中使用table组件 原理&#xff1a; 创建实例时 数组在vue中没有被监听到&#xff0c;属于非响应式数据&#xff0c;数组的下标变化无法监听到 解决方式&#xff1a; <e…...

合宙Air724UG LuatOS-Air script lib API--record

record Table of Contents record record.start(seconds, cbFnc, type, quality, rcdType, format, streamRptLen) record.stop(cbFnc) record.getFilePath() record.getData(offset, len) record.getSize() record.delete() record.exists() record.isBusy() record 模块功能&…...

基于Vgg16和Vgg19深度学习网络的步态识别系统matlab仿真

目录 1.算法运行效果图预览 2.算法运行软件版本 3.部分核心程序 4.算法理论概述 5.算法完整程序工程 1.算法运行效果图预览 2.算法运行软件版本 MATLAB2022A 3.部分核心程序 ................................................................ % 设置训练选项options …...

Java分布式微服务3——Docker

文章目录 Docker介绍安装DockerDocker基础操作Docker服务的启动镜像命令容器命令1. 从docker hub去查看Nginx容器的运行命令2. 查看所有容器状态3. 查看容器日志4. 进入Nginx容器执行命令&#xff0c;修改Html内容&#xff0c;添加“Hello World”5. 停止与开始容器6. 删除容器…...

js字符串替换

在JavaScript中&#xff0c;字符串替换 有多种方法&#xff0c;下面介绍其中一些比较常用的方法。 使用replace()方法、 replace()方法用于在字符串中查…...

网络防御(2)

1. 什么是防火墙&#xff1f; 2. 状态防火墙工作原理&#xff1f; 3. 防火墙如何处理双通道协议&#xff1f; 一、什么是防火墙&#xff1f; 防火墙是一种网络安全设备或软件&#xff0c;用于保护计算机网络免受未经授权的访问&#xff0c;并管理网络流量。它作为一个安全边界…...

[RCTF2019]DontEatMe

前言 一道迷宫题&#xff0c;但是输入被加密后使用&#xff0c;迷宫也需要在程序中找出并没有直接输出 分析 反调试 发现有两个比较特殊的地方&#xff0c;随机数和创建了新线程&#xff0c;随机数后面又被重新赋值给覆盖了&#xff0c;暂时不用管&#xff0c;ZwSetInformat…...

6. CSS(三)

目录 一、盒子模型 &#xff08;一&#xff09;网页布局的本质 &#xff08;二&#xff09;盒子模型组成 &#xff08;三&#xff09;边框&#xff08;border&#xff09; &#xff08;四&#xff09;表格的细线边框 &#xff08;五&#xff09;内边距&#xff08;padding…...

计算机网络—HTTP

这里写目录标题 HTTP是什么HTTP常见状态码HTTP常见字段GET与POST的区别Get和Post是安全和幂等吗PUT幂等&#xff0c;不安全DELETE幂等&#xff0c;不是安全 HTTP缓存技术HTTP缓存实现技术 HTTP1.0优缺点和性能HTTP1.1优缺点和性能HTTP2优缺点和性能HTTP3优缺点和性能HTTP和HTTP…...

Tomcat线程池原理

1. 一个 SpringBoot 项目能同时处理多少请求&#xff1f;tomcat容器&#xff0c; 200 次。 2. 怎么来的&#xff1f; 而点击这些线程&#xff0c;查看其堆栈消息&#xff0c;可以看到 Tomcat、threads、ThreadPoolExecutor 等关键字 基于“短时间内有 200 个请求被立马处理…...

踩坑 视觉SLAM 十四讲第二版 ch13 编译及运行问题

一、安装Geset 库 sudo apt-get install libgtest-dev cd /usr/src/gtest sudo mkdir build cd build sudo cmake .. //一定要以sudo的方式运行&#xff0c;否则没有写入权限 sudo make //这个也一样要以sudo的方式 sudo cp libgtest*.a /usr/local/lib //将生成…...

【设计模式】-装饰器模式

Java 设计模式之装饰器模式 前言 在软件开发中&#xff0c;经常有需求对已有的对象进行功能的扩展&#xff0c;但是传统的继承方式会导致类的数量快速增多&#xff0c;且难以维护。为了解决这个问题&#xff0c;装饰器模式应运而生。 装饰器模式是一种结构型设计模式&#xff…...

七月学习总结

一晃暑期七月份已经结束了&#xff0c;八月份需要做的事情更多。 在成长的路上不断地迷茫&#xff0c;不断地前进。到底才能完成对自己地救赎。 目前想的就是以后走软件开发&#xff0c;往架构方向做&#xff0c;主语言Java或者go&#xff0c;408基础一定要扎实&#xff0c;计…...

Camunda 7.x 系列【6】Spring Boot 集成 Camunda 7.19

有道无术,术尚可求,有术无道,止于术。 本系列Spring Boot 版本 2.7.9 本系列Camunda 版本 7.19.0 源码地址:https://gitee.com/pearl-organization/camunda-study-demo 文章目录 1. 前言2. Camunda Platform Run3. Spring Boot 版本兼容性4. 集成 Spring Boot5. 启动项目…...

Kubernetes —调度器配置

目录 配置文件 扩展点 调度插件 多配置文件 应用于多个扩展点的插件 调度程序配置迁移 你可以通过编写配置文件&#xff0c;并将其路径传给 kube-scheduler 的命令行参数&#xff0c;定制 kube-scheduler 的行为。 调度模板&#xff08;Profile&#xff09;允许你配置 k…...

【微信小程序】申请蓝牙、位置和数据库等相关权限

在小程序的app.json文件中配置requiredPermissions字段&#xff0c;并在其中添加相应的权限。 以下是一个示例app.json文件的配置&#xff0c;包括了蓝牙、位置和数据库等权限的申请&#xff1a; {"pages": ["pages/index/index"],"requiredPermiss…...

ORB-SLAM2学习笔记6之D435i双目IR相机运行ROS版ORB-SLAM2并发布位姿pose的rostopic

文章目录 0 引言1 D435i相机配置2 新增发布双目位姿功能2.1 新增d435i_stereo.cc代码2.2 修改CMakeLists.txt2.3 新增配置文件D435i.yaml 3 编译运行和结果3.1 编译运行3.2 结果3.3 可能出现的问题 0 引言 ORB-SLAM2学习笔记1已成功编译安装ROS版本ORB-SLAM2到本地&#xff0c…...

【数据结构与算法——TypeScript】哈希表

【数据结构与算法——TypeScript】 哈希表(HashTable) 哈希表介绍和特性 哈希表是一种非常重要的数据结构&#xff0c;但是很多学习编程的人一直搞不懂哈希表到底是如何实现的。 在这一章节中&#xff0c;我门就一点点来实现一个自己的哈希表。通过实现来理解哈希表背后的原理…...

【JavaEE】-- HTTP

1. HTTP是什么&#xff1f; HTTP&#xff08;全称为"超文本传输协议"&#xff09;是一种应用非常广泛的应用层协议&#xff0c;HTTP是基于TCP协议的一种应用层协议。 应用层协议&#xff1a;是计算机网络协议栈中最高层的协议&#xff0c;它定义了运行在不同主机上…...

Qt Widget类解析与代码注释

#include "widget.h" #include "ui_widget.h"Widget::Widget(QWidget *parent): QWidget(parent), ui(new Ui::Widget) {ui->setupUi(this); }Widget::~Widget() {delete ui; }//解释这串代码&#xff0c;写上注释 当然可以&#xff01;这段代码是 Qt …...

将对透视变换后的图像使用Otsu进行阈值化,来分离黑色和白色像素。这句话中的Otsu是什么意思?

Otsu 是一种自动阈值化方法&#xff0c;用于将图像分割为前景和背景。它通过最小化图像的类内方差或等价地最大化类间方差来选择最佳阈值。这种方法特别适用于图像的二值化处理&#xff0c;能够自动确定一个阈值&#xff0c;将图像中的像素分为黑色和白色两类。 Otsu 方法的原…...

跨链模式:多链互操作架构与性能扩展方案

跨链模式&#xff1a;多链互操作架构与性能扩展方案 ——构建下一代区块链互联网的技术基石 一、跨链架构的核心范式演进 1. 分层协议栈&#xff1a;模块化解耦设计 现代跨链系统采用分层协议栈实现灵活扩展&#xff08;H2Cross架构&#xff09;&#xff1a; 适配层&#xf…...

2025盘古石杯决赛【手机取证】

前言 第三届盘古石杯国际电子数据取证大赛决赛 最后一题没有解出来&#xff0c;实在找不到&#xff0c;希望有大佬教一下我。 还有就会议时间&#xff0c;我感觉不是图片时间&#xff0c;因为在电脑看到是其他时间用老会议系统开的会。 手机取证 1、分析鸿蒙手机检材&#x…...

Python如何给视频添加音频和字幕

在Python中&#xff0c;给视频添加音频和字幕可以使用电影文件处理库MoviePy和字幕处理库Subtitles。下面将详细介绍如何使用这些库来实现视频的音频和字幕添加&#xff0c;包括必要的代码示例和详细解释。 环境准备 在开始之前&#xff0c;需要安装以下Python库&#xff1a;…...

ios苹果系统,js 滑动屏幕、锚定无效

现象&#xff1a;window.addEventListener监听touch无效&#xff0c;划不动屏幕&#xff0c;但是代码逻辑都有执行到。 scrollIntoView也无效。 原因&#xff1a;这是因为 iOS 的触摸事件处理机制和 touch-action: none 的设置有关。ios有太多得交互动作&#xff0c;从而会影响…...

mysql已经安装,但是通过rpm -q 没有找mysql相关的已安装包

文章目录 现象&#xff1a;mysql已经安装&#xff0c;但是通过rpm -q 没有找mysql相关的已安装包遇到 rpm 命令找不到已经安装的 MySQL 包时&#xff0c;可能是因为以下几个原因&#xff1a;1.MySQL 不是通过 RPM 包安装的2.RPM 数据库损坏3.使用了不同的包名或路径4.使用其他包…...

零知开源——STM32F103RBT6驱动 ICM20948 九轴传感器及 vofa + 上位机可视化教程

STM32F1 本教程使用零知标准板&#xff08;STM32F103RBT6&#xff09;通过I2C驱动ICM20948九轴传感器&#xff0c;实现姿态解算&#xff0c;并通过串口将数据实时发送至VOFA上位机进行3D可视化。代码基于开源库修改优化&#xff0c;适合嵌入式及物联网开发者。在基础驱动上新增…...

使用SSE解决获取状态不一致问题

使用SSE解决获取状态不一致问题 1. 问题描述2. SSE介绍2.1 SSE 的工作原理2.2 SSE 的事件格式规范2.3 SSE与其他技术对比2.4 SSE 的优缺点 3. 实战代码 1. 问题描述 目前做的一个功能是上传多个文件&#xff0c;这个上传文件是整体功能的一部分&#xff0c;文件在上传的过程中…...