-
大数据（9h）FlinkSQL之Lookup Join
概述
- lookup join通常是 查询外部系统的数据 来 充实FlinkSQL的主表
例如：事实表 关联 维度表，维度表在外部系统（如MySQL） - 要求：
1个表具有处理时间属性（基于Processing Time Temporal Join语法）
语法上，和一般JOIN比较，多了FOR SYSTEM_TIME AS OF
另1个表由连接器（a lookup source connector）支持 - Lookup Cache
默认情况下，不启用Lookup Cache
可设置lookup.cache.max-rows和lookup.cache.ttl参数来启用
启用Lookup Cache后，Flink会先查询缓存，缓存未命中才查询外部数据库
启用缓存可加快查询速，但缓存中的记录未必是最新的
SQL参数 说明 connector连接器，可以是 jdbc、kafka、filesystem…driver数据库驱动 lookup.cache.ttlLookup Cache中每行数据 的 最大 存活时间 lookup.cache.max-rowsLookup Cache中的最大行数 当 缓存的行数>
lookup.cache.max-rows时，将清除存活时间最久的记录
缓存中的行 的存货时间 超过lookup.cache.ttl也会被清除pom.xml
环境：WIN10+IDEA+JDK1.8+Flink1.13+MySQL8
<properties> <maven.compiler.source>8maven.compiler.source> <maven.compiler.target>8maven.compiler.target> <flink.version>1.13.6flink.version> <scala.binary.version>2.12scala.binary.version> <slf4j.version>2.0.3slf4j.version> <log4j.version>2.17.2log4j.version> <fastjson.version>2.0.19fastjson.version> <lombok.version>1.18.24lombok.version> <mysql.version>8.0.31mysql.version> properties> <dependencies> <dependency> <groupId>org.apache.flinkgroupId> <artifactId>flink-javaartifactId> <version>${flink.version}version> dependency> <dependency> <groupId>org.apache.flinkgroupId> <artifactId>flink-streaming-java_${scala.binary.version}artifactId> <version>${flink.version}version> dependency> <dependency> <groupId>org.apache.flinkgroupId> <artifactId>flink-clients_${scala.binary.version}artifactId> <version>${flink.version}version> dependency> <dependency> <groupId>org.apache.flinkgroupId> <artifactId>flink-runtime-web_${scala.binary.version}artifactId> <version>${flink.version}version> dependency> <dependency> <groupId>org.apache.flinkgroupId> <artifactId>flink-table-planner-blink_${scala.binary.version}artifactId> <version>${flink.version}version> dependency> <dependency> <groupId>org.apache.flinkgroupId> <artifactId>flink-streaming-scala_${scala.binary.version}artifactId> <version>${flink.version}version> dependency> <dependency> <groupId>org.apache.flinkgroupId> <artifactId>flink-csvartifactId> <version>${flink.version}version> dependency> <dependency> <groupId>org.apache.flinkgroupId> <artifactId>flink-jsonartifactId> <version>${flink.version}version> dependency> <dependency> <groupId>org.apache.flinkgroupId> <artifactId>flink-connector-jdbc_${scala.binary.version}artifactId> <version>${flink.version}version> dependency> <dependency> <groupId>mysqlgroupId> <artifactId>mysql-connector-javaartifactId> <version>${mysql.version}version> dependency> <dependency> <groupId>org.slf4jgroupId> <artifactId>slf4j-apiartifactId> <version>${slf4j.version}version> dependency> <dependency> <groupId>org.slf4jgroupId> <artifactId>slf4j-log4j12artifactId> <version>${slf4j.version}version> dependency> <dependency> <groupId>org.apache.logging.log4jgroupId> <artifactId>log4j-to-slf4jartifactId> <version>${log4j.version}version> dependency> <dependency> <groupId>com.alibabagroupId> <artifactId>fastjsonartifactId> <version>${fastjson.version}version> dependency> <dependency> <groupId>org.projectlombokgroupId> <artifactId>lombokartifactId> <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
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
MySQL建表
DROP DATABASE IF EXISTS db0; CREATE DATABASE db0;- 1
- 2
CREATE TABLE db0.tb0 ( a VARCHAR(255) PRIMARY KEY, b INT(3), c BIGINT(5), d FLOAT(3,2), e DOUBLE(4,2), f DATE DEFAULT '2022-10-24', g TIMESTAMP DEFAULT CURRENT_TIMESTAMP);- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
INSERT db0.tb0 (a,b,c,d,e) VALUES ('aa',1,11,1.11,11.11), ('bb',2,22,2.22,22.22), ('cc',3,33,3.33,33.33); SELECT * FROM db0.tb0;- 1
- 2
- 3
- 4
- 5
对应Flink的建表SQL
SQL
CREATE TEMPORARY TABLE temp_tb0 ( a STRING, b INT, c BIGINT, d FLOAT, e DOUBLE, f DATE, g TIMESTAMP, PRIMARY KEY(a) NOT ENFORCED) WITH ( 'lookup.cache.max-rows' = '2', 'lookup.cache.ttl' = '30 second', 'connector' = 'jdbc', 'driver' = 'com.mysql.cj.jdbc.Driver', 'url' = 'jdbc:mysql://localhost:3306/db0', 'username' = 'root', 'password' = '123456', 'table-name' = 'tb0' )- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
测试代码
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); //创建表，连接MySQL表 tbEnv.executeSql("CREATE TEMPORARY TABLE temp_tb0 (\n" + " a STRING,\n" + " b INT,\n" + " c BIGINT,\n" + " d FLOAT,\n" + " e DOUBLE,\n" + " f DATE,\n" + " g TIMESTAMP,\n" + " PRIMARY KEY(a) NOT ENFORCED)\n" + "WITH (\n" + " 'lookup.cache.max-rows' = '2',\n" + " 'lookup.cache.ttl' = '30 second',\n" + " 'connector' = 'jdbc',\n" + " 'driver' = 'com.mysql.cj.jdbc.Driver',\n" + " 'url' = 'jdbc:mysql://localhost:3306/db0',\n" + " 'username' = 'root',\n" + " 'password' = '123456',\n" + " 'table-name' = 'tb0'\n" + ")"); //执行查询，打印 tbEnv.sqlQuery("SELECT * FROM temp_tb0").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
测试结果打印
+----+----+---+----+------+-------+------------+----------------------------+ | op | a | b | c | d | e | f | g | +----+----+---+----+------+-------+------------+----------------------------+ | +I | aa | 1 | 11 | 1.11 | 11.11 | 2022-10-24 | 2022-11-29 14:57:50.000000 | | +I | bb | 2 | 22 | 2.22 | 22.22 | 2022-10-24 | 2022-11-29 14:57:50.000000 | | +I | cc | 3 | 33 | 3.33 | 33.33 | 2022-10-24 | 2022-11-29 14:57:50.000000 | +----+----+---+----+------+-------+------------+----------------------------+- 1
- 2
- 3
- 4
- 5
- 6
- 7
Lookup Join
FlinkSQL
SELECT * FROM v JOIN t FOR SYSTEM_TIME AS OF v.y ON v.x=t.a- 1
- 2
- 3
- 4
完整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> d = env.addSource(new ManualSource()); Table tb = tbEnv.fromDataStream(d, $("x"), $("y").proctime()); tbEnv.createTemporaryView("v", tb); //创建右表（维度表） tbEnv.executeSql("CREATE TEMPORARY TABLE t ( " + " a STRING, " + " b INT, " + " c BIGINT, " + " d FLOAT, " + " e DOUBLE, " + " f DATE, " + " g TIMESTAMP, " + " PRIMARY KEY(a) NOT ENFORCED) " + "WITH ( " + " 'lookup.cache.max-rows' = '2', " + " 'lookup.cache.ttl' = '30 second', " + " 'connector' = 'jdbc', " + " 'driver' = 'com.mysql.cj.jdbc.Driver', " + " 'url' = 'jdbc:mysql://localhost:3306/db0', " + " 'username' = 'root', " + " 'password' = '123456', " + " 'table-name' = 'tb0' " + ")"); //执行查询，打印 tbEnv.sqlQuery("SELECT * FROM v " + "JOIN t " + " FOR SYSTEM_TIME AS OF v.y " + " ON v.x=t.a").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() {} } }- 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
测试结果
- lookup join通常是 查询外部系统的数据 来 充实FlinkSQL的主表
-
相关阅读:
【Y 码力】WAL 与性能
05-RDD五大特性
MySQL InnoDB 引擎底层解析（二）
【一生一芯】Chap.0 IC常用网站论坛门户 & 如何提出一个技术问题 并尝试解决 | 提问的智慧
【好书推荐】如何快速入门短视频？带你3天成为短视频与Vlog剪辑高手
docker-compose多服务器部署kafka集群
SpringBoot 整合 WebSocket 实现长连接，将数据库中的数据进行推送
Android EventBus 事件订阅/发布框架
python读取视频长度
MySQL8高级_读写分离和分库分表
- 原文地址：https://blog.csdn.net/Yellow_python/article/details/128070761
