connect的使用

上篇：union的使用

connect的使用

可以将两个流不一致的类型连接一起
可以共享状态

直接上代码

package cn._51doit.flink.day03;import org.apache.flink.configuration.Configuration;
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;
/*** connect的使用* 程序操作：* nc -l 8888的窗口输入字符串，将会转成大写输出；* nc -l 8888的窗口输入数字，将会运算输出*/
public class ConnectDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());DataStreamSource<String> lines1 = env.socketTextStream("Master", 8888);SingleOutputStreamOperator<Integer> lines2 = env.socketTextStream("Master", 9999).map(s -> Integer.parseInt(s));//将两个流connect【连接】ConnectedStreams<String, Integer> connected = lines1.connect(lines2);SingleOutputStreamOperator<String> result = connected.map(new CoMapFunction<String, Integer, String>() {//对第一个流进行map运算的方法@Overridepublic String map1(String value) throws Exception {return value.toUpperCase();}//对第二个流进行map运算的方法@Overridepublic String map2(Integer value) throws Exception {return (value * 10) + "";}//这两个流的map方法执行完的返回值会放到新的流中});result.print();env.execute();}}

connect的使用相关推荐

Docker使用遇到问题Got permission denied while trying to connect to the Docker daemon socket解决方案
Got permission denied while trying to connect to the Docker daemon socket at unix:///var/run/docker. ...
Can 't connect to local MySQL server through socket '/tmp/mysql.sock '(2)
安装了mysql, 使用命令mysql -u root -p 弹出Can 't connect to local MySQL server through socket '/tmp/mysql.soc ...
oracle 10035 err 942,案例:Oracle日志报错 Fatal NI connect error 12170 TNS-12535 TNS-00505
天萃荷净 Oracle数据库alert日志文件报错Fatal NI connect error 12170,通过查看mos相关文章找到解决办法今天在一台服务器的日志文件中,发现如下信息: Fatal ...
socket connect阻塞和非阻塞处理
建立socket后默认connect()函数为阻塞连接状态,在大多数实现中,connect的超时时间在75s至几分钟之间,想要缩短超时时间,可解决问题的两种方法:方法一.将socket句柄设置为非阻塞 ...
【Qt】Qt再学习（十六）：QObject::connect: Cannot queue arguments of type ‘QString‘
1.问题描述跨线程使用信号和槽时,如果是非const的引用传参,就会报如下的错误: QObject::connect: Cannot queue arguments of type 'QString ...
【网络编程】非阻塞connect详解
一.为什么使用非阻塞connect TCP连接的建立涉及一个在三路握手过程,阻塞的connect一直等到客户收到自己的SYN的ACK才返回,这需要至少一个RTT时间,RTT时间波动很大从几毫秒到几秒. ...
pyqt designer connect无响应_如何用PyQt编写桌面程序，创建并打开播放列表？
由于电脑上的短视频太多了,并且分别存放在各个子目录下,每次更新后想要整理视频比较麻烦,因此想用Python编写程序来辅助管理视频文件.其实写个Python脚本程序即可实现大器的需求,但为了多练习PyQ ...
Fastlane 入门实战教程从打包到上传iTunes connect
有关神器 Fastlane 持续集成\部署的文章网上挺多,本文定位是入门教程,针对 iOS 应用的持续部署,只需一条命令就可实现从 Xcode 项目到编译\打包\构建\提交审核文章稍微有点长,涵盖 ...
1.3 Quick Start中 Step 7: Use Kafka Connect to import/export data官网剖析（博主推荐）
不多说,直接上干货! 一切来源于官网 http://kafka.apache.org/documentation/ Step 7: Use Kafka Connect to import/export ...
netbackup错误之can not connect on socket(25)
rhel5.5上安装netbackup 7.0,这个版本只能安装在64位系统上.安装完netbackup 7.0后,发现登录界面一直报java认证失败,查看了下日志文件,报如下内容: 查了下系统设置, ...

connect的使用

connect的使用

connect的使用相关推荐

最新文章

热门文章