Flink的Java DataStream API实现多流转换_为伊戏诸侯

未知 02-07 563

1 多流转换 1.1 多流转换简述

在 Flink 的实际应用中，可能需要将不同来源的数据连接合并在一起处理，也有可能需要将一条流拆分成多条流的处理场景，也就是所谓的多流转换。

多流转换：简单划分可以分为 “分流” 和 “合流” 两大类。

分流：一般通过侧输出流（side output）方式实现。合流：可以调用 union、connect、join、coGroup等方式进行连接合并。

本文主要验证内容如下：

分流：侧输出流（Side Output）合流：联合流（Union）、连接流（Connect）。 1.2 项目构建

项目完整目录结构： pom.xml文件：

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://·piler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <flink.version>1.14.4</flink.version> <scala.binary.version>2.12</scala.binary.version> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.12</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> </dependencies> <build> <plugins> <plugin> <artifactId>maven-compiler-plugin</artifactId> <version>3.6.0</version> <configuration> <source>1.8</source> <target>1.8</target> </configuration> </plugin> </plugins> </build> </project>

2 分流 2.1 分流简述

分流：就是将一条数据流拆分成完全独立的两条或者多条数据输出流，也就是基于一个DataStream，得到完全平等的多个子DataStream。

2.2 案例说明数据输入流：定义不同用户（张三、李四等）的登录日志数据数据转换算子：定义一些筛选条件，使用侧输出流区分不同用户的登录日志数据。数据输出流：张三输出流、李四输出流、其余用户输出流。 2.3 代码编写

UserLoginLog.java 类：

package cn.mfox.stream.split.sideout; /** * 用户登录日志 * * @author hy * @version 1.0 * @date 2022/4/15 10:23 */ public class UserLoginLog { /** * 用户名 */ private String name; /** * 登录时间 */ private String time; public UserLoginLog() { } public UserLoginLog(String name, String time) { this.name = name; this.time = time; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getTime() { return time; } public void setTime(String time) { this.time = time; } @Override public String toString() { return "{" + "name='" + name + '\'' + ", time='" + time + '\'' + '}'; } }

SpliteUserLoginLogStreamBySideOutputTag.java 类：

package cn.mfox.stream.split.sideout; 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 hy * @version 1.0 * @date 2022/4/15 9:26 */ public class SpliteUserLoginLogStreamBySideOutputTag { public static void main(String[] args) throws Exception { // 1. 获取环境 StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment(); environment.setParallelism(1); // 2. 添加元素 DataStreamSource<UserLoginLog> loginLogDataStreamSource = environment.fromElements( new UserLoginLog("zhangsan", "2022-04-14 14:14:14"), new UserLoginLog("lisi", "2022-04-14 18:18:18"), new UserLoginLog("wangwu", "2022-04-14 20:20:20"), new UserLoginLog("zhangsan", "2022-04-15 15:15:15"), new UserLoginLog("zhangsan", "2022-04-15 16:16:16"), new UserLoginLog("zhaoliu", "2022-04-15 18:18:18") ); // 3. 处理stream流 SingleOutputStreamOperator<UserLoginLog> processedStream = loginLogDataStreamSource.process( new ProcessFunction<UserLoginLog, UserLoginLog>() { @Override public void processElement(UserLoginLog value, Context ctx, Collector<UserLoginLog> out) throws Exception { if ("zhangsan".equals(value.getName())) { ctx.output(ZhangSanOutTag, value); } else if ("lisi".equals(value.getName())) { ctx.output(LiSiOutTag, value); } else { out.collect(value); } } }); // 4. 获取侧输出流并打印 processedStream.getSideOutput(ZhangSanOutTag).print("ZhangSan Output Stream"); processedStream.getSideOutput(LiSiOutTag).print("LiSi Output Stream"); processedStream.print("Other Output Stream"); // 5. 启动任务 environment.execute(); } /** * 张三输出标签 */ private static OutputTag<UserLoginLog> ZhangSanOutTag = new OutputTag<UserLoginLog>("zhangsan") { }; /** * 李四输出标签 */ private static OutputTag<UserLoginLog> LiSiOutTag = new OutputTag<UserLoginLog>("lisi") { }; } 2.4 运行结果

控制台打印：不同用户的输出流截图：

3 合流

既然一条流可以分开，自然多条数据流也可以合并，在实际应用中，我们经常会遇到来源不同的多条数据流，需要将他们的数据进行合流操作。

3.1 联合流（Union） 3.1.1 联合流简述

联合流（Union）：最简单的合流操作，就是直接将多条数据流合并在一起。联合流特性：

union()方法中的参数可以是多个DataStream，可以实现多条数据流合并。联合操作要求所有的数据类型必须相同。合并之后的新流中会包含所有流中的元素，数据类型不变。 3.1.2 案例说明数据输入流：奇数数据流和偶数数据流使用 union() 方法联合两种数据流并输出。 3.1.3 代码编写

UnionDataStream.java 类：

package cn.mfox.stream.merge.union; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; /** * union联合流（奇数和偶数联合输出） * 1. 不同流之间数据类型必须完全一致。 * 2. union参数可以多个 * * @author hy * @version 1.0 * @date 2022/4/13 15:01 */ public class UnionDataStream { public static void main(String[] args) throws Exception { // 创建环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 奇数流 DataStreamSource<Integer> oddNumStream = env.fromElements(1, 3, 5, 7); // 偶数流 DataStreamSource<Integer> evenStream = env.fromElements(2, 4, 6, 8); // 奇数流联合（union）偶数流 DataStream<Integer> unionDataStream = oddNumStream.union(evenStream); // 打印控制台 unionDataStream.print("UnionDataStream Sink"); // 启动任务 env.execute("UnionDataStream Job"); } } 3.1.4 运行结果

3.2 连接流（Connect） 3.2.1 连接流简述

连接流（Connect）：联合流方式虽然简单，不过受限于数据类型不能改变，灵活性较低，所以实际应用较少，Flink中还有一种连接流（Connect），这种操作就是直接把两条数据流像接线一样连接起来。

连接流特性：

connect()方式中的参数只有一个即只能两个流之间进行连接。两条数据流的数据类型可以不相同。 3.2.2 案例说明数据输入流：数字数据流和字符数据流使用 connect() 方法连接两种数据流并输出。 3.2.3 代码编写

ConnectDataStream.java类：

package cn.mfox.stream.merge.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; /** * 连接流 * 1. connect的参数只能有一个流，即只能两个流之间connect连接 * 2. 两个数据流的数据类型可以不一致 * * @author hy * @version 1.0 * @date 2022/4/13 15:12 */ public class ConnectDataStream { public static void main(String[] args) throws Exception { // 创建环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); // 定义数字流 DataStreamSource<Integer> numStream = env.fromElements(1, 2, 3, 4); // 定义字符串流 DataStreamSource<String> strStream = env.fromElements("A", "B", "C", "D"); // numStream 连接 strStream ConnectedStreams<Integer, String> connectedStreams = numStream.connect(strStream); // map转换 SingleOutputStreamOperator<String> mapStream = connectedStreams.map( new CoMapFunction<Integer, String, String>() { /** * 针对于 numStream 做map转换 * @param value * @return * @throws Exception */ @Override public String map1(Integer value) throws Exception { return "数字：" + value.toString(); } /** * 针对于 strStream 做map转换 * @param value * @return * @throws Exception */ @Override public String map2(String value) throws Exception { return "字符：" + value; } } ); // 控制台打印 mapStream.print("ConnectDataStream Print Sink"); // 启动任务 env.execute("ConnectDataStream Job"); } } 3.2.4 运行结果

4 小结

当前进展：

针对多流转换的简单案例已经初步完成。

后续任务：

多流转换还需进一步学习及使用。后续补充多流join方式。多流转换涉及时间窗口等概念需进一步学习及使用。