flink算子split、select、connect、union、project示例代码

浅浅的花香味﹌ 2022-11-22 12:52 92阅读 0赞

###  对flink相关算子的总结归纳 ###

参考： [flink1.10官方文档][flink1.10]

[github大数据flink知识点][github_flink]

当前maven配置为：

<properties>
         <flink.version>1.10.0</flink.version>
    </properties>
    
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-core</artifactId>
                <version>${flink.version}</version>
            </dependency>
    
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-runtime-web_2.11</artifactId>
                <version>${flink.version}</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_2.11</artifactId>
                <version>${flink.version}</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.xerial.snappy</groupId>
                        <artifactId>snappy-java</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
    
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_2.11</artifactId>
                <version>${flink.version}</version>
            </dependency>
        </dependencies>

以下为代码示例和介绍：

import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.api.java.tuple.Tuple3;
    import org.apache.flink.streaming.api.collector.selector.OutputSelector;
    import org.apache.flink.streaming.api.datastream.ConnectedStreams;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SplitStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.co.CoMapFunction;
    
    import java.util.ArrayList;
    import java.util.List;
    
    
    public class DataTransfroms {
    
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            split(env);
            connect(env);
            union(env);
            project(env);
            env.execute("flink");
        }
    
    }

### \* Connect \[DataStream,DataStream → ConnectedStreams\] ###

Connect 操作用于连接两个或者多个类型不同的 DataStream ，其返回的类型是 ConnectedStreams ，此时被连接的多个 DataStreams 可以共享彼此之间的数据状态。  
但是需要注意的是由于不同 DataStream 之间的数据类型是不同的，如果想要进行后续的计算操作，还需要通过 CoMap 或 CoFlatMap 将 ConnectedStreams 转换回 DataStream：

public static void connect(StreamExecutionEnvironment env) {
            DataStreamSource<Tuple2<String, Integer>> streamSource01 = env.fromElements(new Tuple2<>("a", 3),
                    new Tuple2<>("b", 5));
            DataStreamSource<Integer> streamSource02 = env.fromElements(2, 3, 9);
            // 使用connect进行连接
            ConnectedStreams<Tuple2<String, Integer>, Integer> connect = streamSource01.connect(streamSource02);
            connect.map(new CoMapFunction<Tuple2<String, Integer>, Integer, Integer>() {
                @Override
                public Integer map1(Tuple2<String, Integer> value) throws Exception {
                    return value.f1;
                }
    
                @Override
                public Integer map2(Integer value) throws Exception {
                    return value;
                }
            }).map(x -> x * 100).print();
        }

### 运行结果： ###

500
     200
     300
     300
     900

### \* Union \[DataStream\* → DataStream\] ###

用于连接两个或者多个元素类型相同的 DataStream 。当然一个 DataStream 也可以与其本生进行连接，此时该 DataStream 中的每个元素都会被获取两次：

public static void union(StreamExecutionEnvironment env) {
            DataStreamSource<Tuple2<String, Integer>> streamSource01 = env.fromElements(new Tuple2<>("a", 1),
                    new Tuple2<>("a", 2));
            DataStreamSource<Tuple2<String, Integer>> streamSource02 = env.fromElements(new Tuple2<>("b", 1),
                    new Tuple2<>("b", 2));
            streamSource01.union(streamSource02);
    
            streamSource01.print();
    
            streamSource01.union(streamSource01, streamSource02);
    
            streamSource01.print();
        }

### 运行结果： ###

(a,2)
     (a,1)
     (a,1)
     (a,2)

### \* Split \[DataStream → SplitStream\]： ###

\- 用于将一个 DataStream 按照指定规则进行拆分为多个 DataStream，需要注意的是这里进行的是逻辑拆分，即 Split 只是将数据贴上不同的类型标签，但最终返回的仍然只是一个 SplitStream；

### \* Select \[SplitStream → DataStream\]： ###

\- 想要从逻辑拆分的 SplitStream 中获取真实的不同类型的 DataStream，需要使用 Select 算子

public static void split(StreamExecutionEnvironment env) {
            DataStreamSource<Integer> streamSource = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8);
            // 标记
            SplitStream<Integer> split = streamSource.split(new OutputSelector<Integer>() {
                @Override
                public Iterable<String> select(Integer value) {
                    List<String> output = new ArrayList<String>();
                    output.add(value % 2 == 0 ? "even" : "odd");
                    return output;
                }
            });
            // 获取偶数数据集
            split.select("even").print();
        }

### 运行结果： ###

2
     6
     4
     8

###    \* project \[DataStream → DataStream\] ###

\- project 主要用于获取 tuples 中的指定字段集

public static void project(StreamExecutionEnvironment env) {
            DataStreamSource<Tuple3<String, Integer, String>> streamSource = env.fromElements(
                    new Tuple3<>("li", 22, "2018-09-23"),
                    new Tuple3<>("ming", 33, "2020-09-23"));
            streamSource.project(0, 2).print();
        }

### 运行结果： ###

(ming,2020-09-23)
     (li,2018-09-23)

[flink1.10]: https://ci.apache.org/projects/flink/flink-docs-release-1.10/zh/dev/stream/operators/
[github_flink]: https://github.com/heibaiying/BigData-Notes/blob/master/notes/Flink_Data_Transformation.md