Flink---侧输出流（SideOutput）

ゞ浴缸里的玫瑰 2023-02-22 13:35 8阅读 0赞

要求：将小于32℃输入到侧输入流，大于32℃输入到主流

import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.ProcessFunction
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
    import org.apache.flink.streaming.api.scala.{ OutputTag, 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 SideOutput { 
      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
          .process(new FreeZingAlert())
    
        processedStream.print("processed data")
        processedStream.getSideOutput(new OutputTag[String]("freezing alert")).print("alter data")
    
        env.execute("test")
      }
    }
    //冰点报警，如果小于32F，输出报警信息到测输出流
    class FreeZingAlert() extends ProcessFunction[SensorReading,SensorReading]{ 
      lazy val alertOutput:OutputTag[String] = new OutputTag[String]("freezing alert")
      override def processElement(i: SensorReading, context: ProcessFunction[SensorReading, SensorReading]
        #Context, collector: Collector[SensorReading]): Unit = { 
        if (i.temperature<32.0) { 
          context.output(alertOutput,"freezing alter for " + i.id)
        }else{ 
          //主流
          collector.collect(i)
    
          }
        }
    
    }

![在这里插入图片描述][20200703201412831.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70]

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