Does Flink automatically checkpoint AggregateFunction's state and how to use AggregatingStateDescriptor?

Question

I am implementing a AggregateFunction to measure the duration between two events after .window(EventTimeSessionWindows.withGap(gap)) . After the second event is processed, the window is closed.

Will flink automatically checkpoint the state of the AggregateFunction so that existing data in the accumulator is not lost from restarting?

Since I am not sure about that. I tried to implement AggregatingState in a RichAggregateFunction: class MyAgg extends RichAggregateFunction<IN, ACC, OUT>

AggregatingState requires AggregatingStateDescriptor. Its constructor has this signature:

            String name,
            AggregateFunction<IN, ACC, OUT> aggFunction,
            Class<ACC> stateType) {

I am very confused by the aggFunction. What should be put here? Isn't it the MyAgg that I am trying to define in the first place?

In what context are you using this AggregateFunction? Is this on a streaming window, or somewhere else? — David Anderson
– David Anderson, Commented Jul 30, 2020 at 21:14

David Anderson · Accepted Answer · 2020-07-31 08:25:02Z

2

An AggregateFunction doesn't have any state. But the aggregating state used in a streaming window (and manipulated by an AggregateFunction) is checkpointed as part of the window's state.

A RichAggregateFunction cannot be used in a window context, and an AggregateFunction cannot have its own state. It's designed this way because if an AggregateFunction were allowed to use a state descriptor to define ValueState, for example, then that state wouldn't be mergeable -- and to keep the Window API reasonably clean, all window state needs to be mergeable (for the sake of session windows).

AggregatingState is something you might use in a KeyedProcessFunction, for example. In that context, you need to define how elements are to be aggregated into the accumulator (i.e., the AggregatingState), which you do with an AggregateFunction.

edited Jul 31, 2020 at 8:25

answered Jul 31, 2020 at 6:19

David Anderson

44.3k4 gold badges41 silver badges73 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

Does Flink automatically checkpoint AggregateFunction's state and how to use AggregatingStateDescriptor?

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related