Flink面试总结`持续更新中........

以前都是自己往小本本上总结, 后来小本本没了, 写到这里吧! 有问题请留言, 先谢过了

我们结合实际案例说明一下flink侧输出的用法，假设我们的需求是实时的从kafka接收生产数据，我们需要对迟到超过一定时长的数据或者不符合业务的数据另行处理：
第一步：定义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")

  }

flink中也提供了专对迟到数据侧输出的方法：sideOutputLateData
使用方式：

 val delayOutputTag = OutputTag[T]("delay-side-output")

 val maindata  =input
       .keyBy()
      .window()
      .sideOutputLateData(delayOutputTag)

val delayDataStream = ,maindata.getSideOutput(delayOutputTag)

参考资料

https://ci.apache.org/projects/flink/flink-docs-release-1.6/dev/stream/side_output.html