How to use State Time-To-Live (TTL) in flink job

Viewed 7

In a Flink job, I read a Kafka stream and apply some joins before saving the data in a database. Kafka topic contains two types of data, so I first join both records, create a single row, and save into database. I need the latest data to be stored in the database.

for example -

{"id":1, "item":"1", "amount":300, "type":"revenue"}
{"id":1, "item":"1", "amount":30, "type":"profit"}

I need the data in the table

id | item | revenue | profit
1  |  1   | 300     | 30

I am creating a createTemporaryView ("tableA") on top of the Kafka source and then applying the below-joining logic

Table tr = tableEnv.sqlQuery("select " +
            " coalesce(a.id, b.id) id," +
            " coalesce(a.item, b.item) item," +
            " a.amount as revenue," +
            " b.amount as profit" +
            " from " +
            " (select * from tableA" +
            " where type='revenue') a" +
            " full outer join " +
            " (select * from tableA" +
            " where type='profit') b" +
            " on a.id=b.id, a.item=b.item");

Lastly, save the data in the table. It is working fine. The flink maintains all the state of the join. If id=1 comes 100 times, then flink maintains 100 entries in its state. As a result, a memory problem will arise

Flink is new to me, do we have any setting (TTL) to ensure that it only maintains the last updated value and removes all previous states? In addition, how to configure rockDB settings to maintain robust storage for this state.

0 Answers
Related