CODEWITHSUNDEEP
apache-flink
Tom
Tom

Reputation: 6342

how flink interacts with MySQL for the temporal join with mysql

I am reading at

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

It is using the MySQL as a lookup table in the temporal table join as

-- 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;

I am wondering about how flink interacts with MySQL and whether there are performance issue at the mysql side for temporal join mysql.

The basic question is how flink works for temporal join with mysql.

Upvotes: 2

Views: 564

Answers (1)

David Anderson
David Anderson

Reputation: 43707

You'll find some relevant details in the docs for the Table / JDBC connector: https://ci.apache.org/projects/flink/flink-docs-stable/docs/connectors/table/jdbc/#features. See especially the section describing the Lookup Cache, which says

JDBC connector can be used in temporal join as a lookup source (aka. dimension table). Currently, only sync lookup mode is supported.

By default, lookup cache is not enabled. You can enable it by setting both lookup.cache.max-rows and lookup.cache.ttl.

The lookup cache is used to improve performance of temporal join the JDBC connector. By default, lookup cache is not enabled, so all the requests are sent to external database. When lookup cache is enabled, each process (i.e. TaskManager) will hold a cache. Flink will lookup the cache first, and only send requests to external database when cache missing, and update cache with the rows returned. The oldest rows in cache will be expired when the cache hit to the max cached rows lookup.cache.max-rows or when the row exceeds the max time to live lookup.cache.ttl. The cached rows might not be the latest, users can tune lookup.cache.ttl to a smaller value to have a better fresh data, but this may increase the number of requests send to database. So this is a balance between throughput and correctness.

Upvotes: 2

Related Questions