[jira] [Created] (FLINK-25662) Access KeyedStateStore in WindowAssigner

["Sun Sheng (Jira)" <ji...@apache.org>]

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)