One of the Flink requirements - sideoutput (Application of side output flow: output the temperature higher than 30 ℃ to the mainstream, and output the temperature lower than 30 ℃ to the side flow)

1. First, let's introduce the output stream on the lower side （SideOutPut）

Most of DataStream API The output of the operator of is a single output , It's a stream of some data type . except split operator , You can divide a stream into multiple streams , These streams have the same data type .process function Of side outputs Function can generate multiple streams , And the data types of these streams can be different . One side output Can be defined as OutputTag[X] object ,X Is the data type of the output stream .process function Can pass Context Objects emit an event to one or more side outputs.

2. The realization of requirements

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

import java.util.Properties

object SideOutPutTest {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    env.enableCheckpointing(1000L)
	
	// use kafka As a data source 
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "hadoop101:9092")
    val data = env.addSource(new FlinkKafkaConsumer[String]("sensor",new SimpleStringSchema(),properties))

    val dataStream = data.map(x => {
      val arr = x.split(",")
      SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).trim.toDouble)
    })

    val highTempStream = dataStream
      .process( new SplitTempProcessor(30.0) )

    highTempStream.print("high")
    highTempStream.getSideOutput(new OutputTag[(String,Long,Double)]("low")).print("low")

    env.execute()
  }
}

class SplitTempProcessor(threshold: Double) extends ProcessFunction[SensorReading,SensorReading] {
  override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, SensorReading]#Context, out: Collector[SensorReading]): Unit = {

    if(value.temperature > threshold){
      // If the current temperature value is greater than 30 Celsius output to the mainstream 
      out.collect(value)
    }else{
      // If not more than 10 Centigrade , Output to side output stream 
      ctx.output(new OutputTag[(String,Long,Double)]("low"),(value.id,value.timestamp,value.temperature))
    }
  }
}

3. Result display

Explain , Why? high and low Different types of output , Because of the custom SplitTempProcessor In the realization of ProcessFunction The output type of the side output stream is customized , Similarly, you can continue to use the sample class without defining . The custom data type is used here to illustrate ProcessFunction It can realize more complex business logic .