Flink Join怎么使用

這篇文章主要講解了“Flink Join怎么使用”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“Flink Join怎么使用”吧！

Join算子：兩個數據流通過內部相同的key分區，將窗口內兩個數據流相同key數據元素計算后，合并輸出（類似于mysql表的inner join操作）

  示例環境

java.version: 1.8.x
flink.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<Tuple3<String, String, Integer>> tuple3List1 = DataSource.getTuple3ToList();
        List<Tuple3<String, String, Integer>> tuple3List2 = Arrays.asList(
                new Tuple3<>("伍七", "girl", 18),
                new Tuple3<>("吳八", "man", 30)
        );

        //Datastream 1
        DataStream<Tuple3<String, String, Integer>> dataStream1 = env.fromCollection(tuple3List1)
                //添加水印窗口,如果不添加，則時間窗口會一直等待水印事件時間，不會執行apply
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner((element, timestamp)->System.currentTimeMillis()));

        //Datastream 2
        DataStream<Tuple3<String, String, Integer>> dataStream2 = env.fromCollection(tuple3List2)
                //添加水印窗口,如果不添加，則時間窗口會一直等待水印事件時間，不會執行apply
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Tuple3<String, String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Integer>>() {
                            @Override
                            public long extractTimestamp(Tuple3<String, String, Integer> element, long timestamp) {
                                return System.currentTimeMillis();
                            }
                        }));

        //Datastream 3
        DataStream<String> newDataStream = dataStream1.join(dataStream2)
                .where(new KeySelector<Tuple3<String, String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple3<String, String, Integer> value) throws Exception {
                        System.out.println("first name:" + value.f0 + ",sex:" + value.f1);
                        return value.f1;
                    }
                })
                .equalTo(new KeySelector<Tuple3<String, String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple3<String, String, Integer> 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, String, Integer>, Tuple3<String, String, Integer>, String>() {
                    @Override
                    public void join(Tuple3<String, String, Integer> first, Tuple3<String, String, Integer> second, Collector<String> 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|18
4> 劉六|girl|32|伍七|girl|18
4> 伍七|girl|18|伍七|girl|18
2> 張三|man|20|吳八|man|30
2> 王五|man|29|吳八|man|30
2> 吳八|man|30|吳八|man|30

感謝各位的閱讀，以上就是“Flink Join怎么使用”的內容了，經過本文的學習后，相信大家對Flink Join怎么使用這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！