Hi Faxian, Thanks for thinking on this new approach. Here are my thoughts:
- In case of event time, although, this approach changes semantics of original list state, it could be a good fit for certain use cases. The main advantage is that it is deterministic in event time. The list should end up always in the same order. - In case of processing time, the time skew might be a problem. If task executor's clock jumps back for some reason or it fails and another TE with shifted clock takes over, this can potentially reorder list elements. If we rather think about the list state as a bag, reordering might be ok but there is also a risk that different elements might end up having the same processing time and rewrite each other. - In general, exploding a storage size is a trade-off to achieve more scalability for list state and should be ok if we do not degrade existing approach. Let's see other opinions. Best, Andrey On Fri, Apr 12, 2019 at 10:34 AM Faxian Zhao <faxianz...@gmail.com> wrote: > Refer from PR#5185, I think we can use Timebased RocksDBListState to > resolve it. > Timebased RocksDBListState store list entries dispersed in rocksdb like > RocksDBMapState. > Key pair: > For the timebased flink inner class like StreamRecord(enable > event/ingestion time), the rocksdb key is > #KeyGroup#Key#Namespace#StreamRecord.getTimestamp(). > Otherwise, the key is current procssing time. > Value pair: > The rocksdb value is the entries which have the same > timestamp(event/ingestion/processing time), like the original > RocksDBListState. > > The ListState.get() implement like > org.apache.flink.contrib.streaming.state.RocksDBMapState#iterator. > Generally, it won't load all entries one time. > > The rocksdb store structure. > -----------Key------------------- --------------------Value--------- > #KeyGroup#Key#Namespace #KeyGroup#Key#Namespace#ts3 (max lexicographically > key) > #KeyGroup#Key#Namespace#ts1value1,value2,value7 > #KeyGroup#Key#Namespace#ts2value4,value6 > #KeyGroup#Key#Namespace#ts3value3,value5 > > > Advantage: > 1. Due to the rocksdb store key with lexicographically order, so the > entries is monotonous by time. It's friendly to event time records > processing. > 2. We can store the max timestamp key in the rocksdb default > key(#KeyGroup#Key#Namespace), then we can reverse iterate the stored list. > 3. For the CountEvictor and TimeEvictor, we can stop the iteration early > instead of read all of them into memory. > 4. This ListState is monotonous by time, we can provide some more methods > for event time records processing. > 5. I think it resolve the ttl issue naturally. > > Disadvantage: > 1. It will add 8 bytes cost to store extended timestamp in key part, and > I'm not good at rocksdb, I don't know the performance affect. > 2. For the event time StreamRecord, it will reorder the entries by event > time. This behavior is not align with other ListState implement. > 3. For other records, the key is useless useless overhead. > 4. If all of the entries have the same timestamp, the store structure is > almost same as the original RocksDBListState. > 5. We can't easily implement remove, size method for ListState yet. > > Implement: > We can abstract a new class which is the parent of Time based > RocksDBListState and RocksDBMapState, but we should modify > InternalLargeListState. > I draft some code for this in PR#7675 >
