[WIP] Dataflow Streaming GroupByKey state multiplexing for small keys #33318
Conversation
…oupByKey state over a fixed number of sharding keys. The number of sharding keys is fixed at 32k.
arunpandianp
force-pushed
the
keyMultiplex
branch
from
5af575c to
ede9dea
Compare
...ava/src/main/java/org/apache/beam/runners/dataflow/internal/StateMultiplexingGroupByKey.java
Outdated
Show resolved
Hide resolved
| // Note that Never trigger finishes *at* GC time so it is OK, and | ||
| // AfterWatermark.fromEndOfWindow() finishes at end-of-window time so it is | ||
| // OK if there is no allowed lateness. | ||
| private static boolean triggerIsSafe(WindowingStrategy<?, ?> windowingStrategy) { |
...ava/src/main/java/org/apache/beam/runners/dataflow/internal/StateMultiplexingGroupByKey.java
Outdated
Show resolved
Hide resolved
| } | ||
|
|
||
| @Override | ||
| public PCollection<KV<K, Iterable<V>>> expand(PCollection<KV<K, V>> input) { |
There was a problem hiding this comment.
do we want to verify that the trigger is of some safe type that just uses processing timer or watermark time to trigger?
For example, after count (beyond 1) will have confusing results if it's supposed to count per key but won't now.
There was a problem hiding this comment.
My understanding is all trigger state are per window and the associated state variables are in the window namespace.
Since the userKey is part of the KeyedWindow and the window namespaces, state for different userKeys won't collide.
If there are custom trigger implementation relying on external state, that might have problems depending on how it accesses state.
after count (beyond 1) will have confusing results if it's supposed to count per key but won't now.
IIUC, this change should be transparent and not change semantics for any triggers/windows. Could you explain more on when the after count semantics can change?
There was a problem hiding this comment.
Discussed offline, I was confusing things and this should work as triggers and state are per-window.
| ByteString key = ((KeyedWindow<?>) w).getKey(); | ||
| try { | ||
|
|
||
| // is it correct to use the pane from Keyed window here? |
There was a problem hiding this comment.
see above, I think we need to make sure we only do this for correct triggers. If that is done, I think it would be safe to use the pane.
|
Run Java Precommit |
|
Run Precommit Java |
This is a POC showing how state multiplexing can work for GroupByKey.
The 4k threshold is currently an arbitrary small value and can be tweaked. The constraint is any state tags should not exceed 64k and keys are now part of the windowed state tags.
Need to cleanup comments and add tests, sending this to share the idea and get initial feedback.
R: @scwhittle