Flink多流处理之join(关联)-Toy模板网

这篇具有很好参考价值的文章主要介绍了Flink多流处理之join(关联)。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

Flink的API中只提供了join的算子,并没有left join或者right join,这里我们就介绍一下join算子的使用,其实join算子底层调用的就是coGroup,具体原理这里就不过多介绍了,如果感兴趣可以看我前面发布的文章Flink多流操作之coGroup.文章来源地址https://www.toymoban.com/news/detail-648967.html

数据源

➜  ~ nc -lk 1111
101,A
102,B
103,C
104,D
105,E
106,F

➜  ~ nc -lk 2222
101,A,男,程序员
102,B,男,程序员
103,C,男,会计
104,D,男,安全工程师
106,K,男,程序员
108,女,本科,人事

代码

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
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;

/**
 * @Author: J
 * @Version: 1.0
 * @CreateTime: 2023/8/10
 * @Description: 多流操作-join
 **/
public class FlinkJoin {
    public static void main(String[] args) throws Exception {
        // 构建流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(3);
        // 数据源1,以socket作为数据源
        DataStreamSource<String> socketStream1 = env.socketTextStream("localhost", 1111);
        SingleOutputStreamOperator<String[]> mapStream1 = socketStream1.map(str -> str.split(",")).returns(new TypeHint<String[]>() {});
        // 数据源2,以socket作为数据源
        DataStreamSource<String> socketStream2 = env.socketTextStream("localhost", 2222);
        SingleOutputStreamOperator<String[]> mapStream2 = socketStream2.map(str -> str.split(",")).returns(new TypeHint<String[]>() {});
        // 关联数据流
        DataStream<String> joinedStream = mapStream1.join(mapStream2)
                .where(arr -> arr[0]) // mapStream1以数组中的第一个字段作为关联字段
                .equalTo(arr -> arr[0]) // mapStream2以数组中的第一个字段作为关联字段
                .window(TumblingProcessingTimeWindows.of(Time.seconds(20))) // 以20秒作为一个窗口
                .apply(new JoinFunction<String[], String[], String>() {
                    // 这里是写关联后的具体逻辑
                    @Override
                    public String join(String[] first, String[] second) throws Exception {
                        String result = first[0] + "," + second[1] + "," + second[2] + "," + second[3];
                        return result;
                    }
                });
        // 打印结果数据
        joinedStream.print();

        env.execute("Flink join");
    }
}

结果
```
3> 103,C,男,会计
2> 106,K,男,程序员
2> 101,A,男,程序员
3> 104,D,男,安全工程师
3> 102,B,男,程序员
```
这个API使用起来还是比较简单的,如果想实现left join或者right join的功能就需要通过coGroup来实现了.