Flink: handling streaming state with a key
1 answer
This can be done using the c state operator KeyedStream
. A KeyedStream
splits all records on a key and ensures that all records with the same key go to the same statement instance and interact with the same state.
In code it looks like this:
val stream: DataStream[(String, Long)] = ???
val sumByKey: DataStream[(String, Long)] = stream
.keyBy(_._1) // key on the first attribute
.map(new SumMapper())
class SumMapper extends RichMapFunction[(String, Long), (String, Long)] {
var sumState: ValueState[Long] = _
override def open(config: Configuration) {
// configure state
val sumDesc: ValueStateDescriptor[Long] =
new ValueStateDescriptor[Long]("sum", classOf[Long])
sumState = getRuntimeContext.getState(sumDesc)
}
override def map(in: (String, Long)): (String, Long) = {
val sum = sumState.value() // get current sum from state
val newSum = sum + in._2 // compute new sum
sumState.update(newSum) // update state
(in._1, newSum) // emit result
}
}
+3
source to share