Flink Join怎么使用

这篇文章主要讲解了"Flink Join怎么使用"，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习"Flink Join怎么使用"吧！

Join算子：两个数据流通过内部相同的key分区，将窗口内两个数据流相同key数据元素计算后，合并输出（类似于mysql表的inner join操作）

示例环境

java.version: 1.8.xflink.version: 1.11.1

示例数据源 （项目码云下载）

Flink 系例 之 搭建开发环境与数据

Join.java

package com.flink.examples.functions;import com.flink.examples.DataSource;import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;import org.apache.flink.api.common.eventtime.WatermarkStrategy;import org.apache.flink.api.common.functions.FlatJoinFunction;import org.apache.flink.api.java.functions.KeySelector;import org.apache.flink.api.java.tuple.Tuple3;import org.apache.flink.streaming.api.TimeCharacteristic;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;import org.apache.flink.streaming.api.windowing.time.Time;import org.apache.flink.util.Collector;import java.time.Duration;import java.util.Arrays;import java.util.List;/** * @Description Join算子：两个数据流通过内部相同的key分区，将窗口内两个数据流相同key数据元素计算后，合并输出（类似于mysql表的inner join操作） */public class Join {    /**     * Flink支持了两种Join：Window Join（窗口连接）和Interval Join（时间间隔连接），本示例演示的为Window Join     * 官方文档：https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/stream/operators/joining.html     */    /**     * 两个数据流集合，对相同key进行内联，分配到同一个窗口下，合并并打印     * @param args     * @throws Exception     */    public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.setParallelism(4);        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//        //watermark 自动添加水印调度时间//        env.getConfig().setAutoWatermarkInterval(200);        List> tuple3List1 = DataSource.getTuple3ToList();        List> tuple3List2 = Arrays.asList(                new Tuple3<>("伍七", "girl", 18),                new Tuple3<>("吴八", "man", 30)        );        //Datastream 1        DataStream> dataStream1 = env.fromCollection(tuple3List1)                //添加水印窗口,如果不添加，则时间窗口会一直等待水印事件时间，不会执行apply                .assignTimestampsAndWatermarks(WatermarkStrategy.>forBoundedOutOfOrderness(Duration.ofSeconds(2))                        .withTimestampAssigner((element, timestamp)->System.currentTimeMillis()));        //Datastream 2        DataStream> dataStream2 = env.fromCollection(tuple3List2)                //添加水印窗口,如果不添加，则时间窗口会一直等待水印事件时间，不会执行apply                .assignTimestampsAndWatermarks(WatermarkStrategy.>forBoundedOutOfOrderness(Duration.ofSeconds(2))                        .withTimestampAssigner(new SerializableTimestampAssigner>() {                            @Override                            public long extractTimestamp(Tuple3 element, long timestamp) {                                return System.currentTimeMillis();                            }                        }));        //Datastream 3        DataStream newDataStream = dataStream1.join(dataStream2)                .where(new KeySelector, String>() {                    @Override                    public String getKey(Tuple3 value) throws Exception {                        System.out.println("first name:" + value.f0 + ",sex:" + value.f1);                        return value.f1;                    }                })                .equalTo(new KeySelector, String>() {                    @Override                    public String getKey(Tuple3 value) throws Exception {                        System.out.println("second name:" + value.f0 + ",sex:" + value.f1);                        return value.f1;                    }                })                .window(TumblingEventTimeWindows.of(Time.seconds(1))                .apply(new FlatJoinFunction, Tuple3, String>() {                    @Override                    public void join(Tuple3 first, Tuple3 second, Collector out) throws Exception {                        out.collect(first.f0 + "|" + first.f1 + "|" + first.f2 + "|" + second.f0 + "|" + second.f1 + "|" + second.f2);                    }                })            ;        newDataStream.print();        env.execute("flink Join job");    }}

打印结果

4> 李四|girl|24|伍七|girl|184> 刘六|girl|32|伍七|girl|184> 伍七|girl|18|伍七|girl|182> 张三|man|20|吴八|man|302> 王五|man|29|吴八|man|302> 吴八|man|30|吴八|man|30

感谢各位的阅读，以上就是"Flink Join怎么使用"的内容了，经过本文的学习后，相信大家对Flink Join怎么使用这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是，小编将为大家推送更多相关知识点的文章，欢迎关注！