[
https://issues.apache.org/jira/browse/FLINK-5544?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16270529#comment-16270529
]
ASF GitHub Bot commented on FLINK-5544:
---------------------------------------
Github user shixiaogang commented on the issue:
https://github.com/apache/flink/pull/3359
Very sorry for the delay. I was engaged at the work in the past months,
making flink capable of the terrible data flows in Singles Day.
RocksDBInternalTimerService is among the improvements done.
But we adopt a very different implementation since the initial
implementation presented here has several problems:
* The initial implementation requires other rocksdb instances than the one
used in RocksDBKeyedStateBackend, which makes the resource configuration very
difficult.
* The snapshotting of RocksDBInternalTimerService here is very inefficient.
Though an asynchronous and incremental implementation is available, it will
duplicate much code in RocksDBKeyedStateBackend.
We address these problem by introducing `SecondaryKeyedState`s which
provide non-keyed access methods to the data inside a key group. Similar to
normal keyed state, secondary keyed states are partitioned in to key groups and
are also stored in the backends. Hence these secondary states can also benefit
from asynchronous and incremental snapshotting in `RocksDBKeyedStateBackend`.
What do you think of the changes ? @StefanRRichter
> 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.4.14#64029)
