The class SelfCleaningTemperatureAlertFunction (val threshold: Double)
2 extends KeyedProcessFunction [String, SensorReading (String, Double,
Double)] {
3
4//the keyed state handle for the last temperature
5 private var lastTempState: ValueState [Double]=
_
6//the keyed state handle for the last registered the timer
7 private var lastTimerState: ValueState [Long]=
_
8
9 override def open (parameters: the Configuration) : Unit={
10//register the state for the last temperature
11 val lastTempDesc=new ValueStateDescriptor [Double] (" lastTemp, "classOf [
Double])
12 lastTempState=getRuntimeContext. GetState [Double] (lastTempDescriptor)
13//register the state for the last timer
14 val lastTimerDesc=new ValueStateDescriptor [Long] (" lastTimer, "classOf [
Long])
15 lastTimerState=getRuntimeContext. GetState (timestampDescriptor)
16}