Flink SQL :用于连接表的内存不足

Question

我有一个经常更新的MySql表。我想为过去20秒内更新的每个id拍摄快照，并将值写入redis。我使用binlog作为流输入，并将数据流转换为Flink表。我运行以下sql。

SELECT id, ts, val
FROM my_tbl
WHERE (id, ts) IN
(
   SELECT id, MAX(ts)
   FROM my_tbl
   GROUP BY TUMBLE(proctime, INTERVAL '20' SECOND), id
)

因为我知道表连接会产生过大的状态大小，所以我将StreamQueryConfig设置如下

qConfig.withIdleStateRetentionTime(Time.seconds(600), Time.seconds(1200));

我运行了一天的任务，得到了out of memory (内存不足)错误。我该如何解决这个问题？

Answer 1

您也可以使用时间窗口联接来解决这个问题，而不是使用配置了空闲状态保留时间的常规联接。

下面的查询应该可以做到这一点。

SELECT id, ts, val
FROM my_tbl m1,
     (SELECT id, MAX(ts), TUMBLE_PROCTIME(proctime, INTERVAL '20' SECOND) as ptime
      FROM my_tbl
      GROUP BY TUMBLE(proctime, INTERVAL '20' SECOND), id) m2
WHERE m1.id = m2.id AND m1.ts = m2.ts ANS
      m1.proctime BETWEEN m2.ptime - INTERVAL '25' SECOND AND m2.ptime

窗口连接谓词(BETWEEN)可确保自动清理状态。由于您使用的处理时间不准确，因此我添加了5秒的松弛时间。