大数据之Flink(11) | 状态变成

﹏ヽ暗。殇╰゛Y 2022-12-17 07:30 141阅读 0赞

## TempChangeWarning ##

import learning.SensorReading
    import org.apache.flink.api.common.functions.{ MapFunction, RichFlatMapFunction, RichMapFunction}
    import org.apache.flink.api.common.state.{ ValueState, ValueStateDescriptor}
    import org.apache.flink.configuration.Configuration
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.util.Collector
    
    //输入数据的样例类
    case class SensorReading(id:String,timestamp:Long,temperature:Double)
    
    object StateTest1 { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        val inputStream = env.socketTextStream("hadoop12", 9999)
        val dataStream = inputStream
          .map(data => { 
            val dataArray = data.split(",")
            SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
          })
    
        val warningStream = dataStream
            .keyBy("id")
            .map(new TempChangeWarning(10.0))
    
        warningStream.print()
    
        env.execute("state test job")
      }
    }
    
    //自定义MapFunction 元组(sensor_id,温度，温度) 1对1输出
    class TempChangeWarning(threshold:Double) extends RichMapFunction[SensorReading,(String,Double,Double)]{ 
      //定义状态变量，上一次的温度值
      private var lastTempState:ValueState[Double] = _
    
      override def open(parameters: Configuration): Unit = { 
        lastTempState = getRuntimeContext.getState(new ValueStateDescriptor[Double]("last-temp",classOf[Double]))
      }
    
      override def map(value: SensorReading): (String, Double, Double) ={ 
        //从状态中取出上次的温度值
        val lastTemp = lastTempState.value()
        //更新状态
        lastTempState.update(value.temperature)
    
        //更当前温度值计算差值，然后跟阈值比较，如果大于就报警
        val diff = (value.temperature - lastTemp).abs
        if (diff > threshold) { 
          (value.id, lastTemp, value.temperature)
        } else { 
          (value.id,0.0,0.0)
        }
      }
    }

## TempChangeWarning ##

import learning.SensorReading
    import org.apache.flink.api.common.functions.{ MapFunction, RichFlatMapFunction, RichMapFunction}
    import org.apache.flink.api.common.state.{ ValueState, ValueStateDescriptor}
    import org.apache.flink.configuration.Configuration
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.util.Collector
    
    //输入数据的样例类
    case class SensorReading(id:String,timestamp:Long,temperature:Double)
    
    object StateTest1 { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        val inputStream = env.socketTextStream("hadoop12", 9999)
        val dataStream = inputStream
          .map(data => { 
            val dataArray = data.split(",")
            SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
          })
    
        val warningStream = dataStream
            .keyBy("id")
            .flatMap(new TempChangeWarningWithFlatmap(10.0))
    
        warningStream.print()
    
        env.execute("state test job")
      }
    }
    
    
    //自定义RichFlatMapFunction,可以输出多个结果
    class TempChangeWarningWithFlatmap(threshold:Double) extends RichFlatMapFunction[SensorReading,(String,Double,Double)] { 
    
      lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("last-temp", classOf[Double]))
    
      override def flatMap(value: SensorReading, out: Collector[(String, Double, Double)]): Unit = { 
        //从状态中取出上次的温度值
        val lastTemp = lastTempState.value()
        //更新状态
        lastTempState.update(value.temperature)
    
        //更当前温度值计算差值，然后跟阈值比较，如果大于就报警
        val diff = (value.temperature - lastTemp).abs
        if (diff > threshold) { 
          out.collect((value.id, lastTemp, value.temperature))
        }
      }
    
    }

## flatMapWithState ##

import learning.SensorReading
    import org.apache.flink.api.common.functions.{ MapFunction, RichFlatMapFunction, RichMapFunction}
    import org.apache.flink.api.common.state.{ ValueState, ValueStateDescriptor}
    import org.apache.flink.configuration.Configuration
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.util.Collector
    
    //输入数据的样例类
    case class SensorReading(id:String,timestamp:Long,temperature:Double)
    
    object StateTest1 { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        val inputStream = env.socketTextStream("hadoop12", 9999)
        val dataStream = inputStream
          .map(data => { 
            val dataArray = data.split(",")
            SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
          })
    
        val warningStream = dataStream
            .keyBy("id")
            .flatMapWithState[(String,Double,Double),Double](
              { 
                case (inputData:SensorReading,None) => (List.empty,Some(inputData.temperature))
                case (inputData:SensorReading,lastTemp:Some[Double])=>{ 
                  val diff =(inputData.temperature - lastTemp.get).abs
                  if (diff > 10.0){ 
                    (List((inputData.id,lastTemp.get,inputData.temperature)),Some(inputData.temperature))
                  }else{ 
                    (List.empty,Some(inputData.temperature))
                  }
                }
              }
            )
    
        warningStream.print()
    
        env.execute("state test job")
      }
    }