Sun Sheng created FLINK-25662:
---------------------------------
Summary: Access KeyedStateStore in WindowAssigner
Key: FLINK-25662
URL: https://issues.apache.org/jira/browse/FLINK-25662
Project: Flink
Issue Type: Improvement
Components: API / DataStream
Affects Versions: 1.14.2
Reporter: Sun Sheng
Fix For: 1.14.3
Currently, when implement our own WindowAssigner, by the WindowAssignerContext,
we can call only one function of getCurrentProcessingTime(), but there's no
function to access the KeyedStateStore, in some cases I think it's necessary.
In a stock trading, the events are controlled by it's status, such as
non-trading, continuous-trading, e.g. and the status is in separated events,
and we want to merge the windows from non-trading to continuous-trading as one
window, by assigning a long-window for non-trading and then merge them for
continuous-trading we can do it(using MergingWindowAssigner).
But for recovery, we have to save the status to the KeyedStateStore, because
the status is in separated events.
At present, we have to add a KeyedProcessFunction before windowing, to save the
status and send the status to the window assigner, it works but the code become
more bloat:
stream.keyBy(...).process(MyProcessFunction).keyBy(...).window(MyWindowAssigner)...
I have tried to modified the Flink source code locally, by adding a function to
WindowAssignerContex and implement it in WindowOperator:
// WindowAssigner.java
public abstract static class WindowAssignerContext {
/** Returns the current processing time. */
public abstract long getCurrentProcessingTime();
public abstract KeyedStateStore globalState();
}
// WindowOperator.java
windowAssignerContext =
new WindowAssigner.WindowAssignerContext() {
@Override
public long getCurrentProcessingTime() {
return internalTimerService.currentProcessingTime();
}
@Override
public KeyedStateStore globalState() {
return WindowOperator.this.getKeyedStateStore();
}
};
The implementation is same as the WindowContext, very simple.
By testing locally everything works fine, and my code becomes more clean.
If it's useful I'm glad to work on it. :)
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
