Блог

Как проверить, что последние данные являются первыми данными, использующими состояние Flink

 override def flatMap(value: myData, out: Collector[(String, Double, Double)]): Unit = {  val lastValue = lastValueState.value()  val diff = (value.item_data-lastValue).abs  if (diff gt; Threshold)  out.collect((value.item_id,lastValue,value.item_data))   lastValueState.update(value.item_data)  }  

ValueState#value() возвращает значение null, если ранее ничего не было сохранено для ключа, который в данный момент находится в области действия (ключ, соответствующий текущему value обрабатываемому).