flink 如何与 MySQL 交互以进行与 mysql 的临时连接

Question

我正在阅读

https://ci.apache.org/projects/flink/flink-docs-release-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

您将在表格/ JDBC连接器的文档中找到一些相关的详细信息:https://ci . Apache . org/projects/flink/flink-docs-stable/docs/connectors/Table/JDBC/# features。请特别参阅描述查找缓存的部分，该部分指出

JDBC 连接器在时态连接中可用作查找源（也称为维度表）。目前，仅支持同步查找模式。

默认情况下，不启用查找缓存。您可以通过设置 lookup.cache.max行和 lookup.cache.ttl 来启用它。

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