在窗口关闭时清除 Flink 状态

Question

这是我的代码。我的问题如下

>

以这种方式清除状态是否正确？

这是使用keyBy的正确方法吗？

//有100万个storeId

 orderStream.keyBy(Order::getStoreId)
     .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(16)))
     .trigger(ContinuousEventTimeTrigger.of(Time.seconds(1)))
     .evictor(TimeEvictor.of(Time.seconds(0), true))
     .process(new ProcessWindowFunction<Order, Object, Long, TimeWindow>() {
         MapState<Long, Long> storeCountState;
         @Override
         public void process(Long storeId, Context context, Iterable<Order> elements, Collector<Object> out) throws Exception {
             long sum = 0L;
             for (Order element : elements) {
                 sum++;
             }
             storeCountState.put(storeId, storeCountState.get(storeId) + sum);
         }

         @Override
         public void open(Configuration parameters) throws Exception {
             super.open(parameters);
             MapStateDescriptor<Long, Long> mapStateDescriptor = new MapStateDescriptor();
             storeCountState = getRuntimeContext().getMapState(mapStateDescriptor);
         }

         @Override
         public void close() throws Exception {
             super.close();
             // I clear state when each window close 
             storeCountState.clear();
         }
     })
     .addSink(new PrintSinkFunction<>());

Answer 1

我认为应该重写< code > public void clear(Context Context)throws Exception { } 函数，而不是< code>close()函数。

文档