什么是侧输出
在flink处理数据流时，我们经常会遇到这样的情况：在处理一个数据源时，往往需要将该源中的不同类型的数据做分割处理，如果使用 filter算子对数据源进行筛选分割的话，势必会造成数据流的多次复制，造成不必要的性能浪费；flink中的侧输出就是将数据流进行分割，而不对流进行复制的一种分流机制。flink的侧输出的另一个作用就是对延时迟到的数据进行处理，这样就可以不必丢弃迟到的数据。
侧输出案例
我们结合实际案例说明一下flink侧输出的用法，假设我们的需求是实时的从kafka接收生产数据，我们需要对迟到超过一定时长的数据另行处理：
定义数据类
case class kafkaData(
id: String, //id
eventTime: Long // 时间时间
) {
val delayTime = System.currentTimeMillis() / 1000 - eventTime
}
object kafkaData {
def apply(str: String): kafkaData = {
val strings = str.split(",")
kafkaData(strings.head, strings(1).toLong)
}
}
flink流处理程序
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 从Kafka获取数据流
val kafkaConsumer = KafkaConsumerSource.getComsumer(
load.getString("kafkaBootstrap"),
"test",
load.getString("kafkaGroupId_weather"),
new SimpleStringSchema
)
kafkaConsumer.setStartFromLatest()
第一步：定义OutputTag
// 定义 OutputTag 侧输出的数据格式可以不应和主流的数据格式一样
val delayOutputTag = OutputTag[String]("delay-side-output")
第二步：使用特定的函数将数据发送到侧输出
使用侧输出时需要使用特定的函数发送数据，具体可以使用一下函数：
- ProcessFunction
- CoProcessFunction
- ProcessWindowFunction
- ProcessAllWindowFunction
val dataSource = env.addSource(kafkaConsumer)
.map(kafkaData.apply(_))
//使用特定的函数将数据发送到侧输出
.process(new ProcessFunction[kafkaData, kafkaData] {
override def processElement(value: kafkaData,
ctx: ProcessFunction[kafkaData, kafkaData]#Context,
out: Collector[kafkaData]) = {
if (value.delayTime < 100) {
// 数据常规输出
out.collect(value)
} else {
// 数据侧输出
ctx.output(delayOutputTag, s"数据 ${value.toString} 迟到了 ：" + value.delayTime + "秒")
}
}
})
// 常规数据处理
dataSource.print()
// 对侧输出的数据处理
dataSource.getSideOutput(delayOutputTag).print()
env.execute("Side Outputs Test")
}
我们往kafka中写入两条数据
1,1560522012
1,1560522698
程序输出结果：对不同分割源的数据做了不同的处理
4> 数据 kafkaData(1,1560522012) 迟到了 ：697秒
3> kafkaData(1,1560522698)
flink中也提供了专对迟到数据侧输出的方法：sideOutputLateData
使用方式：
val delayOutputTag = OutputTag[T]("delay-side-output")
val maindata =input
.keyBy(<key selector>)
.window(<window assigner>)
.sideOutputLateData(delayOutputTag)
val delayDataStream = ,maindata.getSideOutput(delayOutputTag)
参考资料
https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/stream/side_output.html
