Flink 侧输出流（SideOutput）

🌸在平时大部分的 DataStream API 的算子的输出是单一输出，也就是某一种或者说某一类数据流，流向相同的地方。

🌸在处理不同的流中，除了 split 算子，可以将一条流分成多条流，这些流的数据类型也都相同。ProcessFunction 的 side outputs 功能可以产生多条流，并且这些流的数据类型可以不一样。一个 side output 可以定义为 OutputTag[X]对象，X 是输出流的数据类型。process function 可以通过 Context 对象发射一个事件到一个或者多个 side outputs。

当使用旁路输出时，首先需要定义一个OutputTag来标识一个旁路输出流

val OutPut=OutputTag[String]("side-output")

注意：OutputTag是如何根据旁路输出流包含的元素类型typed的    

 ✨可以通过以下几种函数发射数据到旁路输出

        ProcessFunction

        CoProcessFunction

        ProcessWindowFunction

        ProcessAllWindowFunction

//将含有特殊字符串的流区分开，数据由两个定义好的工具类向Kafka灌入不同内容的数据，
//然后通过侧输出流(SideOutput)将不同的流进行分离，得到不同的输出import com.alibaba.fastjson.JSON
import com.tech.bean.Person_t
import com.tech.util.KafkaSourceUtil
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collectorobject sideOutputPerson_t {def main(args: Array[String]): Unit = {// UI地址访问：http://localhost:8081/#/job/runningval env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration())val ksu = new KafkaSourceUtil("person_t", "test-consumer-group")val dstream = env.addSource(ksu.getSouceInfo())// 首先需要定义一个OutputTag来标识一个旁路输出流val outputTag = new OutputTag[String]("person_t_side-output")val mainDataStream = dstream.map(line => {JSON.parseObject(line, classOf[Person_t])})val sideOutput = mainDataStream.process(new ProcessFunction[Person_t, String] {override def processElement(value: Person_t,ctx: ProcessFunction[Person_t, String]#Context,out: Collector[String]): Unit = {if (!value.getName.contains("_side")) {out.collect(value.toString)} else {// 测输出流输出的部分ctx.output(outputTag, "sideOutput-> 带有_side标识的数据名称" + value.getName)}}})val sideOutputStream: DataStream[String] = sideOutput.getSideOutput(outputTag)// 测输出流处理sideOutputStream.print("测输出流")// 常规数据处理sideOutput.print("常规数据")env.execute("outSideput")}
}