Flink：配置多个数据源和多个数据源并行处理

港控/mmm° 2024-03-23 13:00 100阅读 0赞

### 需求 ###

根据id获取name

id.txt

1
    2
    3
    4

id\_to\_name.txt

1,zhangsan
    2,lisi
    3,wangwu
    4,zhaoliu

### 依赖 ###

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.7.2</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>1.7.2</version>
        
    </dependency>

### 代码 ###

package com.lagou.edu;
    
    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.CoFlatMapFunction;
    import org.apache.flink.util.Collector;
    
    import java.util.HashMap;
    import java.util.Map;
    
    public class FlinkMultiSource {
    
        public static void main(String[] args) {
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            String inputId = "C:\\data\\input\\id.txt";
            String inputId2Name = "C:\\data\\input\\id_to_name.txt";
            DataStreamSource<String> ids = env.readTextFile(inputId);
    
            DataStreamSource<String> id2Name = env.readTextFile(inputId2Name);
    
            // CoFlatMapFunction
            // 第一个参数 为 ids的数据类型
            // 第二个参数 为 id2Name的数据类型
            // 第三个参数 为 处理完后输出的数据类型（自定义）
            SingleOutputStreamOperator<String> streamOperator = ids.connect(id2Name).flatMap(new CoFlatMapFunction<String, String, String>() {
    
                private Map<String, String> mapId2Name = new HashMap<>();
    
                @Override
                public void flatMap1(String id, Collector<String> collector) throws Exception {
                    // 1
                    String name = mapId2Name.get(id);
                    if (name != null) {
                        collector.collect(name);
                    }
                }
    
                @Override
                public void flatMap2(String idAndName, Collector<String> collector) throws Exception {
                    // 1,张三
                    String[] split = idAndName.split(",");
                    mapId2Name.put(split[0], split[1]);
                }
            });
    
            streamOperator.print();
        }
    }