Flink系列文档-(YY05)-Flink编程API-多流算子
2024-04-12 00:51:02
1 多流连接 connect
connect连接（DataStream,DataStream→ConnectedStreams)
connect翻译成中文意为连接，可以将两个数据类型一样也可以类型不一样DataStream连接成一个新的ConnectedStreams。需要注意的是，connect方法与union方法不同，虽然调用connect方法将两个流连接成一个新的ConnectedStreams，但是里面的两个流依然是相互独立的，这个方法最大的好处是可以让两个流共享State状态，状态相关的内容在后面章节讲解
DataStreamSource<String> ds1 = see.fromElements("a", "b", "c", "d");DataStreamSource<Integer> ds2 = see.fromElements(1, 2, 3, 4, 5, 6);ConnectedStreams<String, Integer> wordAndNumber = ds1.connect(ds2);对ConnectedStreams调用map方法时需要传入CoMapFunction函数:
该接口需要指定3个泛型：
- 第一个输入DataStream的数据类型
- 第二个输入DataStream的数据类型
- 返回结果的数据类型。
- 该接口需要重写两个方法：
- map1方法，是对第1个流进行map的处理逻辑。
- map2方法，是对2个流进行map的处理逻辑
这两个方法必须是相同的返回值类型。指定的输出的数据类型一致.
package com.blok;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;/*** @Date: 22.11.8* @Author: Hang.Nian.YY* @qq: 598196583* @Tips: 学大数据 ,到多易教育* @Description:*/
public class _9Base_API_ConnectFunction{public static void main(String[] args) throws Exception {Configuration conf = new Configuration();conf.setInteger("rest.port", 8888);StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);see.setParallelism(1);DataStreamSource<String> ds1 = see.fromElements("a", "b", "c", "d");DataStreamSource<Integer> ds2 = see.fromElements(1, 2, 3, 4, 5, 6);ConnectedStreams<String, Integer> wordAndNumber = ds1.connect(ds2);// 针对 ConnectedStreams 以后调用的方法 传入的是 CoXXXFunction函数SingleOutputStreamOperator<String> connectMaped = wordAndNumber.map(new CoMapFunction<String, Integer, String>() {// 针对字符串 处理的是左边流的数据@Overridepublic String map1(String value) throws Exception {return null;}// 针对字符串 处理的是右边流的数据@Overridepublic String map2(Integer value) throws Exception {return null;}});see.execute("连接算子") ;}
}
对ConnectedStreams调用flatMap方法,调用flatMap方法，传入的Function是CoFlatMapFunction；
这个接口要重写两个方法：
- flatMap1方法，是对第1个流进行flatMap的处理逻辑；
- flatMap2方法，是对2个流进行flatMap的处理逻辑；
这两个方法都必须返回是相同的类型。
/*** @Date: 22.11.8* @Author: Hang.Nian.YY* @qq: 598196583* @Tips: 学大数据 ,到多易教育* @Description:*/
public class _9Base_API_ConnectFunction02 {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();conf.setInteger("rest.port", 8888);StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);see.setParallelism(1);DataStreamSource<String> ds1 = see.fromElements("a b c d", "e f g h", "j k l", "o p l a");DataStreamSource<String> ds2 = see.fromElements("1 2 3 4 5" , "6 7 8 9 10");ConnectedStreams<String, String> connectedStreams = ds1.connect(ds2);connectedStreams.flatMap(new CoFlatMapFunction<String, String, String>() {@Overridepublic void flatMap1(String value, Collector<String> out) throws Exception {String[] split = value.split("\\s+");for (String word : split) {out.collect(word);}}@Overridepublic void flatMap2(String value, Collector<String> out) throws Exception {String[] split = value.split("\\s+");for (String word : split) {out.collect(word);}}}) ;see.execute("连接算子") ;}
}
2 多流合并
该方法可以将两个或者多个数据类型一致的DataStream合并成一个DataStream。DataStream<T> union(DataStream<T>… streams)可以看出DataStream的union方法的参数为可变参数，即可以合并两个或多个数据类型一致的DataStream。
下面的例子是使用fromElements生成两个DataStream，一个是基数的，一个是偶数的，然后将两个DataStream合并成一个DataStream。
/*** @Date: 22.11.8* @Author: Hang.Nian.YY* @qq: 598196583* @Tips: 学大数据 ,到多易教育* @Description:*/
public class _10Base_API_Union {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();conf.setInteger("rest.port", 8888);StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);see.setParallelism(1);DataStreamSource<Integer> odd = see.fromElements(1, 3, 5, 7, 9);DataStreamSource<Integer> even = see.fromElements(2, 4, 6, 8, 10);// 将两个流合并在一起DataStream<Integer> union = odd.union(even);union.print("所有的数据: ");see.execute("合并union算子");}
}
3 分流操作 - 测流输出
以下function函数，支持将特定数据输出到侧流中：凡是process的函数都有测流输出
- ProcessFunction
- KeyedProcessFunction
- CoProcessFunction
- KeyedCoProcessFunction
- ProcessWindowFunction
- ProcessAllWindowFunction
/*** @Date: 22.11.8* @Author: Hang.Nian.YY* @qq: 598196583* @Tips: 学大数据 ,到多易教育* @Description:*/
public class _11Base_API_SideOut {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();conf.setInteger("rest.port", 8888);StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);see.setParallelism(1);DataStreamSource<YY2> ds = see.fromElements(new YY2(1, "DY", "NM_BT", 100),new YY2(2, "XY", "NM_BT", 100),new YY2(3, "HH", "SD_HZ", 10),new YY2(4, "XH", "SD_HZ", 12));final OutputTag<YY2> sideOut = new OutputTag<YY2>("not good"){};// 使用测流 将不及格的那家伙和优秀的分开// process方法支持测流输出SingleOutputStreamOperator<YY2> processed = ds.process(new ProcessFunction<YY2, YY2>() {@Overridepublic void processElement(YY2 value, ProcessFunction<YY2, YY2>.Context ctx, Collector<YY2> out) throws Exception {if (value.getScore() < 60) { // 将指定规则的不及格的用户 输出到测流ctx.output(sideOut, value);} else { // 将及格的用户输出到主流中 [你们本来就不是一个世界的人, 就不应该有交集]out.collect(value);}}});DataStream<YY2> sideOutput = processed.getSideOutput(sideOut);sideOutput.print("测流输出:-->不优秀的你:") ;processed.print("主流数据:-->优秀的你:") ;see.execute("连接算子");}
}4 协同分组
两个流按照指定的属性分别分组 ,将分组后的数据放在一起处理
package com.blok;import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
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.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;/*** @Date: 22.11.8* @Author: Hang.Nian.YY* @qq: 598196583* @Tips: 学大数据 ,到多易教育* @Description: coGroup 协同分组* 将两个流按照特定的规则进行分组 , 两个流相同的组数据关联*/
public class _12Base_API_Cogroup {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();conf.setInteger("rest.port", 8888);StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);see.setParallelism(1);/*** 1 加载原始数据*/// 输入的数据格式 id,nameDataStreamSource<String> ds1 = see.socketTextStream("linux01", 8899);// 输入的数据格式是 id,event,cityDataStreamSource<String> ds2 = see.socketTextStream("linux01", 9988);/*** 2 处理加载的数据成元组*///id,nameSingleOutputStreamOperator<Tuple2<String, String>> users = ds1.map(line -> {String[] arr = line.split(",");return Tuple2.of(arr[0], arr[1]);}).returns(new TypeHint<Tuple2<String, String>>() {});// id,event,citySingleOutputStreamOperator<Tuple3<String, String, String>> events = ds2.map(line -> {String[] arr = line.split(",");return Tuple3.of(arr[0], arr[1], arr[2]);}).returns(new TypeHint<Tuple3<String, String, String>>() {});//利用coGroup算子，来实现两个流的数据按id相等进行窗口关联（包含inner ，left， right， outer）DataStream<String> res = users.coGroup(events).where(tp -> tp.f0).equalTo(tp -> tp.f0).window(TumblingProcessingTimeWindows.of(Time.seconds(10)))//只要在窗口函数之后才有apply算子.apply(new CoGroupFunction<Tuple2<String, String>, Tuple3<String, String, String>, String>() {@Overridepublic void coGroup(Iterable<Tuple2<String, String>> users, Iterable<Tuple3<String, String, String>> events, Collector<String> out) throws Exception {/*** 实现 left join*/for (Tuple2<String, String> user : users) {boolean flag = false;for (Tuple3<String, String, String> event : events) {out.collect(user.f0 + "," + user.f1 + "," + event.f0 + "," + event.f1 + "," + event.f2);flag = true;}//说明没有事件if (!flag) {out.collect(user.f0 + "," + user.f1 + ",null,null,null");}}}});res.print("left_join") ;see.execute() ;}
}
5 join算子
用于关联两个流（类似于sql中join) , 需要指定join的条件；需要在窗口中进行关联后的逻辑计算；
package com.blok;import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
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.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;/*** @Date: 22.11.8* @Author: Hang.Nian.YY* @qq: 598196583* @Tips: 学大数据 ,到多易教育* @Description: join 两个流进行关联*/
public class _13Base_API_Join {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();conf.setInteger("rest.port", 8888);StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);see.setParallelism(1);/*** 1 加载原始数据*/// 输入的数据格式 id,nameDataStreamSource<String> ds1 = see.socketTextStream("linux01", 8899);// 输入的数据格式是 id,event,cityDataStreamSource<String> ds2 = see.socketTextStream("linux01", 9988);/*** 2 处理加载的数据成元组*///id,nameSingleOutputStreamOperator<Tuple2<String, String>> users = ds1.map(line -> {String[] arr = line.split(",");return Tuple2.of(arr[0], arr[1]);}).returns(new TypeHint<Tuple2<String, String>>() {});// id,event,citySingleOutputStreamOperator<Tuple3<String, String, String>> events = ds2.map(line -> {String[] arr = line.split(",");return Tuple3.of(arr[0], arr[1], arr[2]);}).returns(new TypeHint<Tuple3<String, String, String>>() {});/*** 使用join算子将两个数据流关联在一起*/DataStream<String> res = users.join(events).where(tp -> tp.f0).equalTo(tp -> tp.f0).window(TumblingProcessingTimeWindows.of(Time.seconds(20))) // 划分处理数据的窗口.apply(new JoinFunction<Tuple2<String, String>, Tuple3<String, String, String>, String>() {@Overridepublic String join(Tuple2<String, String> t1, Tuple3<String, String, String> t2) throws Exception {return t1.f0 + "," + t1.f1 + "," + t2.f0 + "," + t2.f1 + "," + t2.f2;}});res.print("join-res: ") ;see.execute() ;}
}
只有同一个窗口中的数据才会触发相对应的数据关联计算
代码中设置不同的窗口类型 ,触发不同的计算时机
// 滚动窗口
// .window(TumblingProcessingTimeWindows.of(Time.seconds(20))) // 划分处理数据的窗口
// .window(TumblingProcessingTimeWindows.of(Time.seconds(30) , Time.seconds(10)))
// 滑动窗口
.window(SlidingProcessingTimeWindows.of(Time.seconds(30), Time.seconds(10)))
//会话窗口
// .window(ProcessingTimeSessionWindows.withGap(Time.seconds(30)))- tumbling window 滚动
DataStream<String> res = users.join(events).where(tp -> tp.f0).equalTo(tp -> tp.f0)// 滚动窗口// .window(TumblingProcessingTimeWindows.of(Time.seconds(20))) // 划分处理数据的窗口.apply(new JoinFunction<Tuple2<String, String>, Tuple3<String, String, String>, String>() {@Overridepublic String join(Tuple2<String, String> t1, Tuple3<String, String, String> t2) throws Exception {return t1.f0 + "," + t1.f1 + "," + t2.f0 + "," + t2.f1 + "," + t2.f2;}});- sliding window滑动
DataStream<String> res = users.join(events).where(tp -> tp.f0).equalTo(tp -> tp.f0)// 滑动窗口.window(SlidingProcessingTimeWindows.of(Time.seconds(30), Time.seconds(10))).apply(new JoinFunction<Tuple2<String, String>, Tuple3<String, String, String>, String>() {@Overridepublic String join(Tuple2<String, String> t1, Tuple3<String, String, String> t2) throws Exception {return t1.f0 + "," + t1.f1 + "," + t2.f0 + "," + t2.f1 + "," + t2.f2;}});- session window 会话
DataStream<String> res = users.join(events).where(tp -> tp.f0).equalTo(tp -> tp.f0)//会话窗口// .window(ProcessingTimeSessionWindows.withGap(Time.seconds(30))).apply(new JoinFunction<Tuple2<String, String>, Tuple3<String, String, String>, String>() {@Overridepublic String join(Tuple2<String, String> t1, Tuple3<String, String, String> t2) throws Exception {return t1.f0 + "," + t1.f1 + "," + t2.f0 + "," + t2.f1 + "," + t2.f2;}});6 广播流
Broadcast State 是 Flink 1.5 引入的新特性。
在开发过程中，如果遇到需要下发/广播配置、规则等低吞吐事件流到下游所有task 时，就可以使用 Broadcast State 特性。下游的 task 接收这些配置、规则并保存为 BroadcastState, 将这些配置应用到另一个数据流的计算中 。
API 介绍 ， 核心要点
- 将需要广播出去的流，调用broadcast方法进行广播转换，得到广播流BroadCastStream
- 然后在主流上调用connect算子，来连接广播流（以实现广播状态的共享处理）
- 在连接流上调用process算子，就会在同一个ProcessFunciton中提供两个方法分别对两个流进行处理，并在这个ProcessFunction内实现“广播状态”的共享
package com.blok;import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;import javax.swing.*;/*** @Date: 22.11.8* @Author: Hang.Nian.YY* @qq: 598196583* @Tips: 学大数据 ,到多易教育* @Description: join 两个流进行关联*/
public class _14Base_API_BroadCast {public static void main(String[] args) throws Exception {Configuration conf = new Configuration();conf.setInteger("rest.port", 8888);StreamExecutionEnvironment see = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);see.setParallelism(1);/*** 1 加载原始数据*/// 输入的数据格式 id,nameDataStreamSource<String> ds1 = see.socketTextStream("linux01", 8899);// 输入的数据格式是 id,event,cityDataStreamSource<String> ds2 = see.socketTextStream("linux01", 9988);/*** 2 处理加载的数据成元组*///id,eventSingleOutputStreamOperator<Tuple2<String, String>> events = ds1.map(line -> {String[] arr = line.split(",");return Tuple2.of(arr[0], arr[1]);}).returns(new TypeHint<Tuple2<String, String>>() {});// id,name,citySingleOutputStreamOperator<Tuple3<String, String, String>> users = ds2.map(line -> {String[] arr = line.split(",");return Tuple3.of(arr[0], arr[1], arr[2]);}).returns(new TypeHint<Tuple3<String, String, String>>() {});/*** 示例代码 :* 流1 用户行为事件 , 一个用户可能有很多不同的行为事件 出现的时间 出现的次数不确定* 流2 用户信息 同一个用户信息数据只会来一次 ,但是来的时间不确定** 将用户信息数据 封装成广播流*///将用户信息数据 封装成广播流MapStateDescriptor<String, Tuple2<String, String>> userInfo = new MapStateDescriptor<>("userInfo", TypeInformation.of(String.class), TypeInformation.of(new TypeHint<Tuple2<String, String>>() {}));BroadcastStream<Tuple3<String, String, String>> broadcast = users.broadcast(userInfo);// 要想使用广播流 , 主流要和广播流进行connectBroadcastConnectedStream<Tuple2<String, String>, Tuple3<String, String, String>> connected = events.connect(broadcast);// 使用 process方法 处理关联了广播流的连接流数据connected.process(new BroadcastProcessFunction<Tuple2<String, String>, Tuple3<String, String, String>, String>() {/*** 处理主流中的数据* @throws Exception*/@Overridepublic void processElement(Tuple2<String, String> value, BroadcastProcessFunction<Tuple2<String, String>, Tuple3<String, String, String>, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {// 从上下文对象中 获取广播状态 这个广播状态是只读的ReadOnlyBroadcastState<String, Tuple2<String, String>> bc = ctx.getBroadcastState(userInfo);if (bc!=null){Tuple2<String, String> user = bc.get(value.f0);if(user!=null){out.collect(value.f0+","+user.f0+","+user.f1+","+value.f1);}else{out.collect(value.f0+",null ,null"+value.f1);}}else{ // 广播变量中没有 用户信息out.collect(value.f0+",null ,null"+value.f1);}}/*** 操作广播数据 ,将广播数据存储在共享状态中* @throws Exception*/@Overridepublic void processBroadcastElement(Tuple3<String, String, String> value, BroadcastProcessFunction<Tuple2<String, String>, Tuple3<String, String, String>, String>.Context ctx, Collector<String> out) throws Exception {// 从上下文对象中 获取广播状态对象 (可以读写的广播流状态)BroadcastState<String, Tuple2<String, String>> bc = ctx.getBroadcastState(userInfo);// 将每条广播数据 存储在广播状态中bc.put(value.f0, Tuple2.of(value.f1, value.f2));}});}
}
Flink系列文档-(YY05)-Flink编程API-多流算子相关推荐
- Flink系列文档-(YY03)-Flink编程基础API-Source
1 Flink编程入口 首先获取flink编程的核心入口对象 /*** 获取批处理入口对象*/// 1) 普通的批处理对象ExecutionEnvironment environment1 = E ...
- Flink系列文档-(YY02)-Flink编程基础-入门示例
1 环境准备-创建项目引入依赖 <dependency><groupId>org.apache.flink</groupId><artifactId>f ...
- Flink系列文档-(YY08)-Flink核心概念
1 核心概念 1.1 基础概念 用户通过算子api所开发的代码,会被flink任务提交客户端解析成jobGraph 然后,jobGraph提交到集群JobManager,转化成ExecutionGra ...
- Flink系列文档-(YY01)-初识Flink
1. 离线批计算与实时流式计算 批计算与流式计算,本质上就是对有界流和无界流的计算 YY-批计算 针对有界流:由于在产出计算结果前可以看到整个(完整)数据集,因而如下计算都可以实现:对数据排序,计 ...
- API是什么？API文档是什么？API和API文档有什么区别？
API是什么? API就是一个接口. 从一个比喻来讲(比喻终究是比喻,但并不是本质):想要将手机上的内容传到电脑上去,可以用一根数据线将二者相连,就可以实现将手机中的内容传到电脑里面去了.其中数据线两 ...
- java openoffice 打印_java调用openoffice将office系列文档转换为PDF的示例方法
前导: 发过程中经常会使用java将office系列文档转换为PDF, 一般都使用微软提供的openoffice+jodconverter 实现转换文档. openoffice既有windows版本也 ...
- 玩转MFC文档视图架构编程1——深入浅出MFC文档/视图架构之基本概念深入浅出MFC文档/视图架构之文档
原创地址: 深入浅出MFC文档/视图架构之基本概念 http://iis.xrtvu.com/Tech/ShowArticle.asp?ArticleID=276 深入浅出MFC文档/视图架构之文档模 ...
- AIO系列文档（2）----TIO使用
AIO系列文档(1)----图解ByteBuffer中介绍了ByteBuffer用法,下面通过介绍t-io介绍如何使用: hello world例子简介 本例子演示的是一个典型的TCP长连接应用,代码 ...
- 老猿学5G扫盲贴：3GPP规范文档命名规则及同系列文档阅读指南
专栏:Python基础教程目录 专栏:使用PyQt开发图形界面Python应用 专栏:PyQt入门学习 老猿Python博文目录 老猿学5G博文目录 在学习5G规范过程中,有些内容把握不定的时候,有时 ...
最新文章
- 【leetcode】394. Decode String
- go使用grpc实现异步_(python、go)基于ETCD的gRPC分布式服务器实现详解
- [android] 百度地图开发 (一).申请AK显示地图及解决显示空白网格问题
- Jdk1.8新特性(一) 接口的default方法
- 自动化测试---页面截图
- python实现GBDT算法的回归、二分类以及多分类，算法流程解读并可视化
- epoch、 iteration和batchsize的区别
- rabbitMq多个队列多种任务同时监听
- 人工智能（腾讯AI）-身份证识别技术使用接口说明
- 简单的权限管理系统——数据库设计和实现
- cmd中的%~dp0
- Chrome远程调试手机端UC浏览器
- 【转】图片热点链接使用方法
- 原创 关于微信拼车小程序开发的需求分析（分析建模）
- 论文笔记：HIE-SQL:History Information Enhanced Network for Context-Dependent Text-to-SQL Semantic Parsing
- /dev/tty、/dev/ttyS/、/dev/ttyUSB区别
- 柠檬班学习性能测试day2_性能测试入门（0519）
- winds操作系统下登录本地MySQL和远程MySQL
- ECMAScript6
- 基于 Java 机器学习自学笔记 （第63-65天：集成学习之AdaBoost）