Flink(process-KeyedProcessFunction和flatmap-RichFlatMapFunction)

一时失言乱红尘 2023-02-22 13:39 5阅读 0赞

需求：温度不能相差10 如果超过10就报警  
下面有两中方法实现 **`process和flatmap`**  
**`process-KeyedProcessFunction`**

import org.apache.flink.api.common.state.{ ValueState, ValueStateDescriptor}
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.util.Collector
    
    
    //传感器读书样列类
    case class SensorReading(id :String,timestamp:Long,temperature:Double)
    
    object ProcessFunctionTest { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        val stream = env.socketTextStream("hadoop102", 7777)
    
        val dataStream = stream.map(data=>{ 
          val dataArray = data.split(",")
          SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
        })
          .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) { 
            override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000
          })
        
        val processedStream = dataStream.keyBy(_.id)
            .process(new TempChangeAlert(10.0))
    
        dataStream.print("input data")
        processedStream.print("processed data")
    
        env.execute()
    
      }
    }
    
    
    class TempChangeAlert(threshold:Double) extends KeyedProcessFunction[String,SensorReading,(String,Double,Double)] { 
      //定义一个状态变量，保存上次的温度值
      lazy val lastTempState:ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp",
        classOf[Double]))
      override def processElement(i: SensorReading, context: KeyedProcessFunction[String, SensorReading,
        (String, Double, Double)]#Context, collector: Collector[(String, Double, Double)]): Unit = { 
        //获取上次的温度值
        val lastTemp = lastTempState.value()
        //用当前的温度值和上次的求差，如果大于值，输出报警信息 abs绝对值
        val diff = (i.temperature - lastTemp).abs
        if (diff > threshold){ 
          collector.collect((i.id,lastTemp,i.temperature))
        }
        lastTempState.update(i.temperature)
      }
    }

![在这里插入图片描述][20200703222501245.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]  
**`flatmap-RichFlatMapFunction`**

import org.apache.flink.api.common.functions.RichFlatMapFunction
    import org.apache.flink.api.common.state.{ ValueState, ValueStateDescriptor}
    import org.apache.flink.api.scala._
    import org.apache.flink.configuration.Configuration
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
    import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.util.Collector
    
    
    //传感器读书样列类
    case class SensorReading(id :String,timestamp:Long,temperature:Double)
    
    object ProcessFunctionTest { 
      def main(args: Array[String]): Unit = { 
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        val stream = env.socketTextStream("hadoop102", 7777)
    
        val dataStream = stream.map(data=>{ 
          val dataArray = data.split(",")
          SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
        })
          .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) { 
            override def extractTimestamp(t: SensorReading): Long = t.timestamp * 1000
          })
    
        val processedStream = dataStream.keyBy(_.id)
            .flatMap(new TempChangeAlert(10.0))
    
    
        dataStream.print("input data")
        processedStream.print("processed data")
    
        env.execute()
    
      }
    }
    class TempChangeAlert(threshold:Double) extends RichFlatMapFunction[SensorReading,(String,Double,Double)] { 
    
      private var lastTempState:ValueState[Double] = _
    
      //用lazy也可以，但是这里我们用open
      override def open(parameters: Configuration): Unit = { 
        //初始化的时候声明state变量
        lastTempState = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp",
          classOf[Double]))
      }
    
      override def flatMap(i: SensorReading, collector: Collector[(String, Double, Double)]): Unit = { 
          //获取上次的温度值
          val lastTemp = lastTempState.value()
          //用当前的温度值和上次的求差，如果大于值，输出报警信息 abs绝对值
          val diff = (i.temperature - lastTemp).abs
          if (diff > threshold){ 
            collector.collect((i.id,lastTemp,i.temperature))
          }
          lastTempState.update(i.temperature)
    
    
      }
    }

[20200703222501245.png]: https://img-blog.csdnimg.cn/20200703222501245.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200703222445190.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1,size_16,color_FFFFFF,t_70