Блог

Я пытаюсь создать точку сохранения (используя API процессора состояний) для функции richcogroup. Поскольку функция CoGroup использует окно, я подумал, что мне нужно написать состояние окна. Из документации я нашел это

 BootstrapTransformation<Account> transformation = OperatorTransformation
    .bootstrapWith(accountDataSet)
    // When using event time windows, it is important
    // to assign timestamps to each record.
    .assignTimestamps(account -> account.timestamp)
    .keyBy(acc -> acc.id)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .reduce((left, right) -> left   right);
 

где вместо сокращения у меня есть

 .apply(WindowFuntion)
 

Но в этом случае, как я могу записать значение и состояние карты моей функции richcogroup, поскольку функция WindowFunction не имеет открытого метода для переопределения?

Спасибо, что уделили мне время.