flink state问题

大家好 

我有一个去重的需求，想节省内存用的bloomfilter，代码如下：

.keyBy(_._1).process(new KeyedProcessFunctionString,(String,String),String {

var state:ValueState[BloomFilter[CharSequence]]= null

override def open(parameters: Configuration): Unit = {

val stateDesc = new ValueStateDescriptor("state",TypeInformation.of(new TypeHintBloomFilter[CharSequence]{}))

state = getRuntimeContext.getState(stateDesc)

}

override def processElement(value: (String, String), ctx: KeyedProcessFunction[String, (String, String), String]#Context, out: Collector[String]) = {

var filter = state.value

if(filter==null){

println("null filter")

filter= BloomFilter.createCharSequence}

//val contains = filter.mightContain(value._2)

if(!filter.mightContain(value._2)) {

filter.put(value._2)

state.update(filter)

out.collect(value._2)

}

}

})

通过日志我看到每次我从savepoint恢复的时候这个state里面的bloomfilter都是null，这是为什么啊

*来自志愿者整理的flink邮件归档