[
https://issues.apache.org/jira/browse/FLINK-5544?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15964122#comment-15964122
]
ASF GitHub Bot commented on FLINK-5544:
---------------------------------------
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/3359
@vpernin Thanks very much for your attention. The PR is supposed to work on
1.3-SNAPSHOT, but it's not testable now due to some known bugs.
Besides, i want to add support for asynchronous snapshots of timers in this
pull request. Currently, the snapshots for timers are taken synchronously ---
no stream record can be processed before the snapshots are taken. In our tests
where there are millions of timers, it takes approximately several seconds to
complete the snapshotting. The performance, hence, is significantly degraded
when the checkpoint frequency is large.
To allow asynchronous snapshotting, we need some refactoring on how
internal timer services are restored and snapshotted. Now
`InternalTimerService` s, similar to keyed states, are stored in
`KeyedStateBackend`. That way, we can benefit from the optimizations made on
the snapshotting of keyed states, taking snapshots asynchronously (and
incrementally in the near future).
I am working on this work right now. It's appreciated that you could help
test the feature when it is done.
> Implement Internal Timer Service in RocksDB
> -------------------------------------------
>
> Key: FLINK-5544
> URL: https://issues.apache.org/jira/browse/FLINK-5544
> Project: Flink
> Issue Type: New Feature
> Components: State Backends, Checkpointing
> Reporter: Xiaogang Shi
> Assignee: Xiaogang Shi
>
> Now the only implementation of internal timer service is
> HeapInternalTimerService which stores all timers in memory. In the cases
> where the number of keys is very large, the timer service will cost too much
> memory. A implementation which stores timers in RocksDB seems good to deal
> with these cases.
> It might be a little challenging to implement a RocksDB timer service because
> the timers are accessed in different ways. When timers are triggered, we need
> to access timers in the order of timestamp. But when performing checkpoints,
> we must have a method to obtain all timers of a given key group.
> A good implementation, as suggested by [~StephanEwen], follows the idea of
> merge sorting. We can store timers in RocksDB with the format
> {{KEY_GROUP#TIMER#KEY}}. In this way, the timers under a key group are put
> together and are sorted.
> Then we can deploy an in-memory heap which keeps the first timer of each key
> group to get the next timer to trigger. When a key group's first timer is
> updated, we can efficiently update the heap.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
