devbox
2023面试高手
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

大数据（9h）FlinkSQL双流JOIN_flink双流join的等待机制

作者：2023面试高手 | 2024-05-30 18:45:37

赞

踩

flink双流join的等待机制

文章目录

重点是Lookup Join和Processing Time Temporal Join，其它随意

1、环境

WIN10+IDEA2021+JDK1.8+本地MySQL8

<properties>
    <maven.compiler.source>8</maven.compiler.source>
    <maven.compiler.target>8</maven.compiler.target>
    <flink.version>1.13.6</flink.version>
    <scala.binary.version>2.12</scala.binary.version>
    <slf4j.version>2.0.3</slf4j.version>
    <log4j.version>2.17.2</log4j.version>
    <fastjson.version>2.0.19</fastjson.version>
    <lombok.version>1.18.24</lombok.version>
</properties>
<!-- https://mvnrepository.com/ -->
<dependencies>
    <!-- Flink -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-runtime-web_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!-- FlinkSQL -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-csv</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-json</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <!-- 日志 -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>${slf4j.version}</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>${slf4j.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-to-slf4j</artifactId>
        <version>${log4j.version}</version>
    </dependency>
    <!-- JSON解析 -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>${fastjson.version}</version>
    </dependency>
    <!-- 简化JavaBean书写 -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>${lombok.version}</version>
    </dependency>
</dependencies>

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83

2、Temporal Joins

2.1、基于处理时间（重点）

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Hi {
    public static void main(String[] args) {
        //创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        //创建流式表执行环境
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env);
        //双流
        DataStreamSource<Tuple2<String, Integer>> d1 = env.fromElements(
                Tuple2.of("a", 2),
                Tuple2.of("b", 3));
        DataStreamSource<P> d2 = env.fromElements(
                new P("a", 4000L),
                new P("b", 5000L));
        //创建临时视图
        tbEnv.createTemporaryView("v1", d1);
        tbEnv.createTemporaryView("v2", d2);
        //双流JOIN
        tbEnv.sqlQuery("SELECT * FROM v1 LEFT JOIN v2 ON v1.f0=v2.pid").execute().print();
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class P {
        private String pid;
        private Long timestamp;
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36

结果

+----+-------+-------+-------------+-------------+
| op |    f0 |    f1 |         pid |   timestamp |
+----+-------+-------+-------------+-------------+
| +I |     a |     2 |      (NULL) |      (NULL) |
| -D |     a |     2 |      (NULL) |      (NULL) |
| +I |     a |     2 |           a |        4000 |
| +I |     b |     3 |      (NULL) |      (NULL) |
| -D |     b |     3 |      (NULL) |      (NULL) |
| +I |     b |     3 |           b |        5000 |
+----+-------+-------+-------------+-------------+

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

2.1.1、设置状态保留时间

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;
import java.util.Scanner;

public class Hi {
    public static void main(String[] args) {
        //创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        //创建流式表执行环境
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env);
        //设置状态保留时间
        tbEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5L));
        //双流
        DataStreamSource<Tuple2<String, Long>> d1 = env.addSource(new AutomatedSource());
        DataStreamSource<String> d2 = env.addSource(new ManualSource());
        //创建临时视图
        tbEnv.createTemporaryView("v1", d1);
        tbEnv.createTemporaryView("v2", d2);
        //双流JOIN
        tbEnv.sqlQuery("SELECT * FROM v1 INNER JOIN v2 ON v1.f0=v2.f0").execute().print();
    }

    /** 手动输入的数据源（请输入a或b进行测试） */
    public static class ManualSource implements SourceFunction<String> {
        public ManualSource() {}

        @Override
        public void run(SourceFunction.SourceContext<String> sc) {
            Scanner scanner = new Scanner(System.in);
            while (true) {
                String str = scanner.nextLine().trim();
                if (str.equals("STOP")) {break;}
                if (!str.equals("")) {sc.collect(str);}
            }
            scanner.close();
        }

        @Override
        public void cancel() {}
    }

    /** 自动输入的数据源 */
    public static class AutomatedSource implements SourceFunction<Tuple2<String, Long>> {
        public AutomatedSource() {}

        @Override
        public void run(SourceFunction.SourceContext<Tuple2<String, Long>> sc) throws InterruptedException {
            for (long i = 0L; i < 999L; i++) {
                Thread.sleep(1000L);
                sc.collect(Tuple2.of("a", i));
                sc.collect(Tuple2.of("b", i));
            }
        }

        @Override
        public void cancel() {}
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63

测试结果

2.2、基于事件时间

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Hello {
    public static void main(String[] args) {
        //创建流和表的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env);
        //创建数据流，设定水位线
        tbEnv.executeSql("CREATE TABLE v1 (" +
                "  x STRING PRIMARY KEY," +
                "  y BIGINT," +
                "  ts AS to_timestamp(from_unixtime(y,'yyyy-MM-dd HH:mm:ss'))," +
                "  watermark FOR ts AS ts - INTERVAL '2' SECOND" +
                ") WITH (" +
                "  'connector'='filesystem'," +
                "  'path'='src/main/resources/a.csv'," +
                "  'format'='csv'" +
                ")");
        tbEnv.executeSql("CREATE TABLE v2 (" +
                "  x STRING PRIMARY KEY," +
                "  y BIGINT," +
                "  ts AS to_timestamp(from_unixtime(y,'yyyy-MM-dd HH:mm:ss'))," +
                "  watermark FOR ts AS ts - INTERVAL '2' SECOND" +
                ") WITH (" +
                "  'connector'='filesystem'," +
                "  'path'='src/main/resources/b.csv'," +
                "  'format'='csv'" +
                ")");
        //执行查询
        tbEnv.sqlQuery("SELECT * " +
                "FROM v1 " +
                "LEFT JOIN v2 FOR SYSTEM_TIME AS OF v1.ts " +
                "ON v1.x = v2.x"
        ).execute().print();
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37

打印结果

+----+---+------------+-------------------------+--------+------------+-------------------------+
| op | x |          y |                      ts |     x0 |         y0 |                     ts0 |
+----+---+------------+-------------------------+--------+------------+-------------------------+
| +I | a | 1666540800 | 2022-10-24 00:00:00.000 | (NULL) |     (NULL) |                  (NULL) |
| +I | b | 1666540803 | 2022-10-24 00:00:03.000 |      b | 1666540802 | 2022-10-24 00:00:02.000 |
| +I | c | 1666540806 | 2022-10-24 00:00:06.000 |      c | 1666540803 | 2022-10-24 00:00:03.000 |
+----+---+------------+-------------------------+--------+------------+-------------------------+

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

3、Lookup Join（重点）

Lookup Join是基于Processing Time Temporal Join的

详细链接：https://yellow520.blog.csdn.net/article/details/128070761

4、Interval Joins（基于间隔JOIN）

SQL

SELECT * FROM v1
LEFT JOIN v2 ON v1.x=v2.x AND
v1.y BETWEEN (v2.y - INTERVAL '2' SECOND) AND (v2.y + INTERVAL '1' SECOND)

  • 1
  • 2
  • 3

Java

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Scanner;

import static org.apache.flink.table.api.Expressions.$;

public class Hi {
    public static void main(String[] args) {
        //创建流和表的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env);
        //创建数据流
        DataStreamSource<String> d1 = env.addSource(new ManualSource());
        DataStreamSource<String> d2 = env.addSource(new AutomatedSource());
        //创建动态表，并声明一个额外的字段来作为处理时间字段
        Table t1 = tbEnv.fromDataStream(d1, $("x"), $("y").proctime());
        Table t2 = tbEnv.fromDataStream(d2, $("x"), $("y").proctime());
        //创建临时视图
        tbEnv.createTemporaryView("v1", t1);
        tbEnv.createTemporaryView("v2", t2);
        //执行查询
        tbEnv.sqlQuery("SELECT * FROM v1 LEFT JOIN v2 ON v1.x=v2.x AND " +
                "v1.y BETWEEN (v2.y - INTERVAL '2' SECOND) AND (v2.y + INTERVAL '1' SECOND)"
        ).execute().print();
    }

    /** 手动输入的数据源 */
    public static class ManualSource implements SourceFunction<String> {
        public ManualSource() {}

        @Override
        public void run(SourceFunction.SourceContext<String> sc) {
            Scanner scanner = new Scanner(System.in);
            while (true) {
                String str = scanner.nextLine().trim();
                if (str.equals("STOP")) {break;}
                if (!str.equals("")) {sc.collect(str);}
            }
            scanner.close();
        }

        @Override
        public void cancel() {}
    }

    /** 自动输入的数据源 */
    public static class AutomatedSource implements SourceFunction<String> {
        public AutomatedSource() {}

        @Override
        public void run(SourceFunction.SourceContext<String> sc) throws InterruptedException {
            for (int i = 0; i < 999; i++) {
                Thread.sleep(801);
                sc.collect("a");
                sc.collect("b");
            }
        }

        @Override
        public void cancel() {}
    }
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66

测试结果

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/2023面试高手/article/detail/648159
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号