Table of Contents |
---|
Status
Current state: Under DiscussionAccepted
Discussion thread: here
JIRA: here
Released: 0.10.1.0
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
The cache has two functions. First, it continues to serve as a read buffer for data that is sent to the state store, just like today. Second, it serves as a write deduplicator for the state store (just like today) as well as for the downstream processor node(s). So the tradeoff is "getting each update, i.e., a low update delay -- but a large CPU and storage load" vs. "skipping some updates, i.e., larger update delay -- but a reduced resource load". Note that this optimization does not affect correctness. The optimization is applicable to aggregations and . It is not applicable to other operators like joins.
The proposal calls for one parameter to be added to the streams configuration options:
...
The semantics of this parameter is that data is forwarded and flushed whenever the earliest of commit.interval.ms (note this parameter already exists and specifies the frequency with which a processor flushes its state) or cache pressure hits. Both parameters specify per-processor buffering time and buffering size. They These are global parameters in the sense that they apply to all processor nodes in the topology, i.e., it will not be possible to specify different parameters for each node. Such fine grained control is probably not needed from the DSL.
Continuing with the motivating example above, what is forwarded for K1, assuming all three operations hit in cache, is a final Change value <V1 + V10 + V100, null>.
For the low-level Processor API, we propose no changes, i.e., this KIP does not apply to the Processor API, just the DSL.
Public Interfaces
This KIP adds the cache.max.bytes.buffering
configuration to the streams configuration as described above.
...
Change the RocksDB implementation for KStream windowed aggregations, to not use “range queries” but multiple gets, so that we can leverage caches for it as wellmerge range queries from the cache with range queries from RocksDb.
Extract the LRU cache out of RocksDBStore, as a separate store for KGroupedStream.aggregate() / reduce(), and KStream.aggregateByKey() / reduceByKey().Add the LRU cache for
KTable.to()
operator.. Forward entires downstream on flush or evict.Adjust context forwarding logic so that forwarding happens on cache flush
Add the above config into StreamsConfig
...
The advantage of this parameter is that it could be smaller than the existing commit.interval.ms
, and thus allow for smaller time intervals spent buffering. However, if desired, commit.interval.ms
could also be set to a smaller value, so it is unclear what the benefit of yet another additional timeout would be.
- Expose this caching to the Processor API:
For the low-level Processor API, we could allow users to enable/disabling caching on a store-by-store basis using a new .enableCaching() call. For example:
TopologyBuilder builder = new TopologyBuilder();
builder.addStateStore(Stores.create("Counts").withStringKeys().withIntegerValues().persistent().enableCaching().build(), "Process");
The implication of adding caching to the processor API is that the caching layer will also do forwarding. Hence if a store's cache is enabled, the use of context.forward would not needed anymore. However, this poses the problem of a user having to change their code (to use context.forward) each time they want to enable or disable caching. As such, more thought is needed for this approach. Furthermore, there are several other ways of doing caching in the processor API and perhaps we do not want to prescribe this one as the only way.