Hello, I am trying to use Flink SQL api to join two tables. My stream data source is kafka (defined through catalog and schema registry) and my enrichment data is located in relational database (JDBC connector). I think this setup reflects quite common use case
Enrichment table definition looks like this: CREATE TABLE dim ( ID BIGINT, ENRICH STRING, FROM_DATE TIMESTAMP(6), TO_DATE TIMESTAMP(6), WATERMARK FOR TO_DATE AS TO_DATE ) WITH ( 'connector' = 'jdbc', 'url' = ‘…’, 'table-name' = ‘…’, 'username' = ‘…’, 'password' = ‘…’ ) And this is join I use against stream coming from kafka (table with watermark spec), trunc is udf: SELECT TRUNC(T1.START_TIME,'HH24') as `START_TIME`, D1.ENRICH as `ENRICH`, T1.FIELD as `FIELD`, FROM `kafka.topic` T1, dim D1 WHERE T1.ENRICH_ID = D1.ID AND T1.START_TIME between D1.TO_DATE - INTERVAL ‘1’ DAY AND D1.TO_DATE AND T1.START_TIME >= D1.FROM_DATE Result job graph contains two table source scan operators together with interval join operator. The problem I am trying to solve is how to change the character of enrichment table. Currently, related operator task reads whole data from table when the job start and finishes afterwards. Ideally, I would like to have continuously updated enrichment table. Is it possible to achieve without CDC for example by querying whole database periodically or use some kind of cache for keys? We can assume that enrichment table is append only, there are no deletes or updates, only inserts for new time intervals If updates are not possible, how can I deal with finished task? Due to a known issue [1], all checkpoints are aborted . Maybe I could live with restarting job to get new enrichment data as it is not refreshed so frequently, but checkpointing is a must. flink version 1.12 regards Marek [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-147%3A+Support+Checkpoints+After+Tasks+Finished
