stateSnapshots по требованию в mapWithState

Question

Я выполняю потоковую передачу данных из Kafka (интервал между партиями 10 секунд), преобразую RDD в PairRDD, а затем сохраняю RDD в состояние с помощью mapWithState(). Ниже приведен код:

    JavaPairDStream<String, Object> transformedStream = stream
            .mapToPair(record -> new Tuple2<>(record.getKey(), record))
            .mapWithState(StateSpec.function(updateDataFuncGDM).numPartitions(32)).stateSnapshots();

transformedStream.foreachRDD(rdd -> {
        //if flag is true, put the RDD to a SQL table, and run a query to do some aggregations liek sum, avg etc
       // if flag is false, return;
        }

Теперь я продолжаю обновлять данные в состоянии, и для определенного события я изменяю флаг на true, и я помещаю эти данные в таблицу, и делаю мои вычисления.

Проблема здесь в том, что, так как я получаю "stateSnapshots" в каждом пакете, это неэффективно, а mapWithState сохраняет много данных в памяти, и по мере роста состояния это станет еще хуже. Кроме того, поскольку mapWithState проверяет данные после каждых 10 итераций, это занимает много времени, потому что данные очень большие.

Я хочу получить stateSnapshot состояния только по требованию (т.е. только в итерации foreachRDD, когда флаг имеет значение true) Но я не нашел много способов поиграться с состоянием