大部分的DataStream API的算子的输出是单一输出，也就是某种数据类型的流。除了split算子，可以将一条流分成多条流，这些流的数据类型也都相同。process function的side outputs功能可以产生多条流，并且这些流的数据类型可以不一样。一个side output可以定义为OutputTag[X]对象，X是输出流的数据类型。process function可以通过Context对象发射一个事件到一个或者多个side outputs。
需求:温度大于30输出一条流和温度小于30输出一条流 代码实现 import learning.SensorReading import org.apache.flink.streaming.api.functions.ProcessFunction import org.apache.flink.streaming.api.scala._ import org.apache.flink.util.Collector object SideOutputTest { 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) }) //用ProcessFunction的测输出流实现分流操作 val highTempStream = dataStream .process(new SplitTempProcessor(30)) val lowTempStream = highTempStream.getSideOutput(new OutputTag[(String,Double,Long)]("low-temp")) //打印输出 highTempStream.print("high") lowTempStream.print("low") env.execute("side output job") } } //自定义ProcessFunction，用于区分高低温度的数据 class SplitTempProcessor(threshold: Int) extends ProcessFunction[SensorReading,SensorReading]{ override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, SensorReading] #Context, out: Collector[SensorReading]): Unit = { //判断当前数据的温度值，如果大于阈值，输出到主流，如果小于阈值，输出到测输出流 if (value.temperature > threshold){ out.collect(value) }else{ ctx.output(new OutputTag[(String,Double,Long)]("low-temp"),(value.id,value.temperature,value.timestamp)) } } }结果
