flink SQL流媒体查询的状态大小和检查点大小与TVF聚合保持增长

发布于 2025-01-24 02:56:58 字数 1310 浏览 4 评论 0原文

我们正在使用Flink SQL来定义流处理管道，该管道在5分钟的固定窗口上计算总和聚集。查询看起来像这样：

 INSERT INTO BigTableTable
 SELECT CONCAT_WS('#', user, attribute, bucket) as rowkey, ROW(sum_attribute), cell_timestamp
 FROM (
   SELECT
     user,
     attribute,
     DATE_FORMAT(window_end, 'yyyy-MM-dd:HH') AS bucket,
     SUM(attribute_to_sum) AS sum_attribute,
     window_end as cell_timestamp
   FROM TABLE(TUMBLE(TABLE inputTable, DESCRIPTOR(event_time), INTERVAL '5' MINUTES))
   GROUP BY user, attribute, window_start, window_end)

我们使用embeddedrocksdbstatebackend，并使用增量检查点以及不一致的检查点。

我们启用了一些本地RockSDB指标来监视现场状态大小，并监视检查站的大小，我们看到这些大小会随着时间的推移而不断增长。这有点违反直觉，因为我们的理解是，窗口关闭并开火后，窗口状态将被扫除。

我们还使用以下方式启用了表API中的状态TTL：tableConvig.getConfig.setIdLestTereTeention（持续时间。minutes.ofminutes（10）），但这似乎并没有改变任何东西。

这是状态大小和检查点大小的两个图：状态实时大小检查点

大小溪流？

更新：

这是带有对齐检查点的两个指标 checkpoint size 状态实时大小

原文

We are using Flink SQL to define a stream processing pipeline that computes a SUM aggregation over a fixed window of 5 minutes. The query looks something like this:

 INSERT INTO BigTableTable
 SELECT CONCAT_WS('#', user, attribute, bucket) as rowkey, ROW(sum_attribute), cell_timestamp
 FROM (
   SELECT
     user,
     attribute,
     DATE_FORMAT(window_end, 'yyyy-MM-dd:HH') AS bucket,
     SUM(attribute_to_sum) AS sum_attribute,
     window_end as cell_timestamp
   FROM TABLE(TUMBLE(TABLE inputTable, DESCRIPTOR(event_time), INTERVAL '5' MINUTES))
   GROUP BY user, attribute, window_start, window_end)

We are using EmbeddedRocksDBStateBackend with incremental checkpointing and also unaligned checkpoints.

We enabled some native rocksdb metrics to monitor the live state size, and also monitor the checkpoint size and we are seeing that these keep growing over time. This is a bit counterintuitive since our understanding is that the window state will be wipped out once the window closes and fires.

We have also enabled the state ttl in table api with: tableEnvironment.getConfig.setIdleStateRetention(Duration.ofMinutes(10)), but this doesn't seem to change anything.

Here are the two graphs for state size and checkpoint size:
State Live Size
Checkpoint Size

Why does the state keep growing and is not bounded by the number of keys in the inout stream?

Update:

Here's the same two metrics with aligned checkpoints Checkpoint Size State Live Size

分享到QQ

分享到微博