Flink:按键处理有状态流

Question

我有数据流，例如带有ID的JSON记录。

我想处理数据，以便具有相同键的所有记录都由同一有状态任务处理。

我该怎么做？

Answer 1

这可以通过KeyedStream上的有状态运算符来完成。KeyedStream对键上的所有记录进行分区，并确保具有相同键的所有记录都转到同一运算符实例并与相同的状态进行交互。

在代码中，这看起来像：

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
}
}