flink如何与MySQL进行临时连接

Question

我正在阅读

https://ci.apache.org/projects/flink/flink文档-释放- 1.13 -/- docs/dev/table/sql/queries/joins/# lookup-join,

在时态表连接中使用MySQL作为查找表

-- Customers is backed by the JDBC connector and can be used for lookup joins
CREATE TEMPORARY TABLE Customers (
id INT,
name STRING,
country STRING,
zip STRING
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://mysqlhost:3306/customerdb',
'table-name' = 'customers'
);
-- enrich each order with customer information
SELECT o.order_id, o.total, c.country, c.zip
FROM Orders AS o
JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
ON o.customer_id = c.id;

我想知道flink是如何与MySQL交互的，以及是否有性能问题在MySQL端临时连接MySQL。

基本的问题是flink是如何与mysql进行临时连接的。

Answer 1

您可以在Table/JDBC连接器的文档中找到一些相关细节:https://ci.apache.org/projects/flink/flink-docs-stable/docs/connectors/table/jdbc/#features。请特别参阅描述查找缓存的部分，其中说

JDBC连接器可以在临时连接中用作查找源(即。维度表)。目前只支持同步查询模式。

默认情况下，不启用查找缓存。您可以通过设置lookup.cache来启用它。Max-rows and lookup.cache.ttl.

查找缓存用于提高JDBC连接器临时连接的性能。默认情况下，未启用查找缓存，因此所有请求都发送到外部数据库。当启用查找缓存时，每个进程(即TaskManager)将持有一个缓存。Flink将首先查找缓存，只有在缓存丢失时才向外部数据库发送请求，并使用返回的行更新缓存。当缓存达到最大缓存行数时，缓存中最老的行将过期。最大行数，或者当行超过存活的最大时间时。缓存的行可能不是最新的，用户可以将lookup.cache.ttl调优到较小的值以获得更好的新数据，但这可能会增加发送到数据库的请求数量。所以这是吞吐量和正确性之间的平衡。