본문으로 건너뛰기
최종 수정: 에

Lookup Join

Lookup Join can optimize the performance of dimension table joins in Flink. When using Flink JDBC Connector for dimension table joins, you may encounter the following problems:

  • Flink JDBC Connector uses synchronous query mode: each time upstream data (such as Kafka) sends a record, the Doris dimension table is queried immediately, leading to high query latency in high-concurrency scenarios.
  • Queries executed via JDBC are usually point lookups one record at a time, while Doris recommends batch queries for better query efficiency.

Using Lookup Join in Flink Doris Connector has the following advantages:

  • Caches upstream data in batches, avoiding the high latency and database pressure caused by per-record queries.
  • Executes association queries asynchronously, increasing data throughput and reducing the Doris query load.

Example​

The dimension table must be configured with jdbc-url, which the Lookup queries go through. Enable the cache with the lookup.cache.* options:

CREATE TABLE fact_table (
`id` BIGINT,
`name` STRING,
`city` STRING,
`process_time` as proctime()
) WITH (
'connector' = 'kafka',
...
);

create table dim_city(
`city` STRING,
`level` INT,
`province` STRING,
`country` STRING
) WITH (
'connector' = 'doris',
'fenodes' = '127.0.0.1:8030',
'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
'table.identifier' = 'dim.dim_city',
'username' = 'root',
'password' = '',
'lookup.cache.max-rows' = '100000',
'lookup.cache.ttl' = '300s'
);

SELECT a.id, a.name, a.city, c.province, c.country, c.level
FROM fact_table a
LEFT JOIN dim_city FOR SYSTEM_TIME AS OF a.process_time AS c
ON a.city = c.city

Options​

KeyDefault ValueRequiredComment
lookup.cache.max-rows-1NThe maximum number of rows cached by lookup. The default value is -1, meaning caching is disabled.
lookup.cache.ttl10sNThe maximum cache time for lookup. The default is 10s.
lookup.max-retries1NThe number of retries after a lookup query failure.
lookup.jdbc.asyncFALSENWhether to enable asynchronous lookup. The default is false.
lookup.jdbc.read.batch.size128NIn asynchronous lookup, the maximum batch size per query.
lookup.jdbc.read.batch.queue-size256NIn asynchronous lookup, the size of the intermediate buffer queue.
lookup.jdbc.read.thread-size3NThe number of jdbc threads for lookup in each task.